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}
203
204impl Default for BulkCallConfig {
205 fn default() -> Self {
206 Self {
207 max_slots_per_call: 10_000,
208 max_targets_per_call: 250,
209 max_concurrent_calls: 4,
210 point_read_threshold: 2,
211 pre_shanghai_extractor: false,
212 dispatch: CallDispatch::PerCall,
213 max_slots_per_request: 25_000,
214 }
215 }
216}
217
218impl BulkCallConfig {
219 fn normalized(self) -> Self {
220 Self {
221 max_slots_per_call: self.max_slots_per_call.max(1),
222 max_targets_per_call: self.max_targets_per_call.max(1),
223 max_concurrent_calls: self.max_concurrent_calls.max(1),
224 max_slots_per_request: self.max_slots_per_request.max(1),
225 ..self
226 }
227 }
228
229 fn extractor(&self) -> Bytes {
230 if self.pre_shanghai_extractor {
231 Bytes::from_static(STORAGE_EXTRACTOR_CODE_PRE_SHANGHAI)
232 } else {
233 Bytes::from_static(STORAGE_EXTRACTOR_CODE)
234 }
235 }
236}
237
238pub fn pack_slots_calldata(slots: &[U256]) -> Bytes {
241 let mut out = Vec::with_capacity(slots.len() * 32);
242 for slot in slots {
243 out.extend_from_slice(&slot.to_be_bytes::<32>());
244 }
245 out.into()
246}
247
248pub fn decode_packed_values(data: &[u8], expected: usize) -> Option<Vec<U256>> {
254 if data.len() != expected * 32 {
255 return None;
256 }
257 Some(data.chunks_exact(32).map(U256::from_be_slice).collect())
258}
259
260pub fn encode_multi_target_calldata(targets: &[(Address, Vec<U256>)]) -> Bytes {
265 let calls: Vec<IMulticall3::Call3> = targets
266 .iter()
267 .map(|(target, slots)| IMulticall3::Call3 {
268 target: *target,
269 allowFailure: true,
270 callData: pack_slots_calldata(slots),
271 })
272 .collect();
273 IMulticall3::aggregate3Call { calls }.abi_encode().into()
274}
275
276pub fn decode_multi_target_response(
279 targets: &[(Address, Vec<U256>)],
280 response: &[u8],
281) -> Vec<(Address, U256, StorageFetchResult<U256>)> {
282 let decoded = match IMulticall3::aggregate3Call::abi_decode_returns(response) {
283 Ok(results) if results.len() == targets.len() => results,
284 Ok(results) => {
285 return per_target_errors(targets, || {
286 StorageFetchError::custom(format!(
287 "aggregate3 returned {} results for {} extraction targets",
288 results.len(),
289 targets.len()
290 ))
291 });
292 }
293 Err(e) => {
294 return per_target_errors(targets, || {
295 StorageFetchError::custom(format!("failed to decode aggregate3 response: {e}"))
296 });
297 }
298 };
299
300 let mut out = Vec::with_capacity(targets.iter().map(|(_, s)| s.len()).sum());
301 for ((target, slots), result) in targets.iter().zip(decoded) {
302 if !result.success {
303 out.extend(slots.iter().map(|slot| {
304 (
305 *target,
306 *slot,
307 Err(StorageFetchError::custom(
308 "extractor subcall failed (allowFailure=true); the target may be a precompile",
309 )),
310 )
311 }));
312 continue;
313 }
314 match decode_packed_values(&result.returnData, slots.len()) {
315 Some(values) => out.extend(
316 slots
317 .iter()
318 .zip(values)
319 .map(|(slot, value)| (*target, *slot, Ok(value))),
320 ),
321 None => out.extend(slots.iter().map(|slot| {
322 (
323 *target,
324 *slot,
325 Err(StorageFetchError::custom(format!(
326 "extractor at {target} returned {} bytes, expected {}",
327 result.returnData.len(),
328 slots.len() * 32
329 ))),
330 )
331 })),
332 }
333 }
334 out
335}
336
337fn per_target_errors(
338 targets: &[(Address, Vec<U256>)],
339 make: impl Fn() -> StorageFetchError,
340) -> Vec<(Address, U256, StorageFetchResult<U256>)> {
341 targets
342 .iter()
343 .flat_map(|(target, slots)| slots.iter().map(|slot| (*target, *slot, Err(make()))))
344 .collect()
345}
346
347#[derive(Debug, Clone, PartialEq, Eq)]
349enum CallPlan {
350 Single { target: Address, slots: Vec<U256> },
352 Multi { targets: Vec<(Address, Vec<U256>)> },
354}
355
356impl CallPlan {
357 fn request_slot_count(&self) -> usize {
358 match self {
359 Self::Single { slots, .. } => slots.len(),
360 Self::Multi { targets } => targets.iter().map(|(_, s)| s.len()).sum(),
361 }
362 }
363}
364
365fn plan_calls(requests: &[(Address, U256)], config: &BulkCallConfig) -> Vec<CallPlan> {
374 let mut order: Vec<Address> = Vec::new();
375 let mut groups: HashMap<Address, Vec<U256>> = HashMap::new();
376 for (address, slot) in requests {
377 groups
378 .entry(*address)
379 .or_insert_with(|| {
380 order.push(*address);
381 Vec::new()
382 })
383 .push(*slot);
384 }
385
386 let mut plans = Vec::new();
387 let mut packable: Vec<(Address, Vec<U256>)> = Vec::new();
388 for address in order {
389 let slots = groups.remove(&address).expect("grouped above");
390 for chunk in slots.chunks(config.max_slots_per_call) {
391 let full = chunk.len() == config.max_slots_per_call;
392 if full || address == MULTICALL3_ADDRESS {
397 plans.push(CallPlan::Single {
398 target: address,
399 slots: chunk.to_vec(),
400 });
401 } else {
402 packable.push((address, chunk.to_vec()));
403 }
404 }
405 }
406
407 let mut current: Vec<(Address, Vec<U256>)> = Vec::new();
409 let mut current_slots = 0usize;
410 let flush =
411 |current: &mut Vec<(Address, Vec<U256>)>, plans: &mut Vec<CallPlan>| match current.len() {
412 0 => {}
413 1 => {
414 let (target, slots) = current.pop().expect("len checked");
415 plans.push(CallPlan::Single { target, slots });
416 }
417 _ => plans.push(CallPlan::Multi {
418 targets: std::mem::take(current),
419 }),
420 };
421 for (address, slots) in packable {
422 let would_overflow = current_slots + slots.len() > config.max_slots_per_call
423 || current.len() >= config.max_targets_per_call;
424 if !current.is_empty() && would_overflow {
425 flush(&mut current, &mut plans);
426 current_slots = 0;
427 }
428 current_slots += slots.len();
429 current.push((address, slots));
430 }
431 flush(&mut current, &mut plans);
432
433 plans
434}
435
436fn overrides_for_plan(plan: &CallPlan, extractor: &Bytes) -> StateOverride {
439 let mut overrides = StateOverride::default();
440 match plan {
441 CallPlan::Single { target, .. } => {
442 overrides.insert(
443 *target,
444 AccountOverride::default().with_code(extractor.clone()),
445 );
446 }
447 CallPlan::Multi { targets } => {
448 overrides.insert(
449 MULTICALL3_ADDRESS,
450 AccountOverride::default().with_code(multicall3_runtime_code().clone()),
451 );
452 for (target, _) in targets {
453 overrides.insert(
454 *target,
455 AccountOverride::default().with_code(extractor.clone()),
456 );
457 }
458 }
459 }
460 overrides
461}
462
463fn plan_call_parts(plan: &CallPlan) -> (Address, Bytes) {
465 match plan {
466 CallPlan::Single { target, slots } => (*target, pack_slots_calldata(slots)),
467 CallPlan::Multi { targets } => (MULTICALL3_ADDRESS, encode_multi_target_calldata(targets)),
468 }
469}
470
471fn decode_plan_response(
473 plan: &CallPlan,
474 bytes: &[u8],
475) -> Vec<(Address, U256, StorageFetchResult<U256>)> {
476 match plan {
477 CallPlan::Single { target, slots } => match decode_packed_values(bytes, slots.len()) {
478 Some(values) => slots
479 .iter()
480 .zip(values)
481 .map(|(slot, value)| (*target, *slot, Ok(value)))
482 .collect(),
483 None => slots
484 .iter()
485 .map(|slot| {
486 (
487 *target,
488 *slot,
489 Err(StorageFetchError::custom(format!(
490 "extractor at {target} returned {} bytes, expected {} — the \
491 provider may not support eth_call state overrides, or the \
492 target is a precompile",
493 bytes.len(),
494 slots.len() * 32
495 ))),
496 )
497 })
498 .collect(),
499 },
500 CallPlan::Multi { targets } => decode_multi_target_response(targets, bytes),
501 }
502}
503
504fn plan_error_results(
506 plan: &CallPlan,
507 err: StorageFetchError,
508) -> Vec<(Address, U256, StorageFetchResult<U256>)> {
509 match plan {
510 CallPlan::Single { target, slots } => slots
511 .iter()
512 .map(|slot| (*target, *slot, Err(err.clone())))
513 .collect(),
514 CallPlan::Multi { targets } => targets
515 .iter()
516 .flat_map(|(target, slots)| {
517 slots.iter().map({
518 let err = err.clone();
519 move |slot| (*target, *slot, Err(err.clone()))
520 })
521 })
522 .collect(),
523 }
524}
525
526async fn execute_plan<P: Provider<AnyNetwork>>(
527 provider: &P,
528 block: BlockId,
529 plan: CallPlan,
530 extractor: &Bytes,
531) -> Vec<(Address, U256, StorageFetchResult<U256>)> {
532 let overrides = overrides_for_plan(&plan, extractor);
533 let (to, data) = plan_call_parts(&plan);
534 let tx = TransactionRequest::default().to(to).input(data.into());
535
536 let response: Result<Bytes, _> = provider
537 .client()
538 .request("eth_call", (tx, block, overrides))
539 .await;
540
541 match response {
542 Ok(bytes) => decode_plan_response(&plan, &bytes),
543 Err(e) => plan_error_results(&plan, StorageFetchError::provider("eth_call", &e)),
544 }
545}
546
547#[derive(Debug, serde::Deserialize)]
550struct CallManyEntry {
551 value: Option<Bytes>,
552 error: Option<serde_json::Value>,
553}
554
555async fn execute_plans_call_many<P: Provider<AnyNetwork>>(
562 provider: &P,
563 number: alloy_eips::BlockNumberOrTag,
564 plans: &[CallPlan],
565 extractor: &Bytes,
566) -> Result<Vec<(Address, U256, StorageFetchResult<U256>)>, StorageFetchError> {
567 let mut overrides = StateOverride::default();
572 let mut transactions = Vec::with_capacity(plans.len());
573 for plan in plans {
574 for (address, account) in overrides_for_plan(plan, extractor) {
575 overrides.insert(address, account);
576 }
577 let (to, data) = plan_call_parts(plan);
578 transactions.push(serde_json::json!({ "to": to, "data": data }));
579 }
580
581 let bundles = serde_json::json!([{ "transactions": transactions }]);
582 let context = serde_json::json!({ "blockNumber": number, "transactionIndex": -1 });
583 let response: Vec<Vec<CallManyEntry>> = provider
584 .client()
585 .request("eth_callMany", (bundles, context, overrides))
586 .await
587 .map_err(|e| StorageFetchError::provider("eth_callMany", &e))?;
588
589 let entries: Vec<CallManyEntry> = response.into_iter().flatten().collect();
590 if entries.len() != plans.len() {
591 return Err(StorageFetchError::custom(format!(
592 "eth_callMany returned {} results for {} bundled calls",
593 entries.len(),
594 plans.len()
595 )));
596 }
597
598 let mut out = Vec::new();
599 for (plan, entry) in plans.iter().zip(entries) {
600 match entry.value {
601 Some(bytes) => out.extend(decode_plan_response(plan, &bytes)),
602 None => {
603 let detail = entry
604 .error
605 .map(|e| e.to_string())
606 .unwrap_or_else(|| "no value returned".to_string());
607 out.extend(plan_error_results(
608 plan,
609 StorageFetchError::custom(format!("eth_callMany transaction failed: {detail}")),
610 ));
611 }
612 }
613 }
614 Ok(out)
615}
616
617fn group_plans_for_call_many(
620 plans: Vec<CallPlan>,
621 max_slots_per_request: usize,
622) -> Vec<Vec<CallPlan>> {
623 let mut requests: Vec<Vec<CallPlan>> = Vec::new();
624 let mut current: Vec<CallPlan> = Vec::new();
625 let mut current_slots = 0usize;
626 for plan in plans {
627 let slots = plan.request_slot_count();
628 if !current.is_empty() && current_slots + slots > max_slots_per_request {
629 requests.push(std::mem::take(&mut current));
630 current_slots = 0;
631 }
632 current_slots += slots;
633 current.push(plan);
634 }
635 if !current.is_empty() {
636 requests.push(current);
637 }
638 requests
639}
640
641pub async fn fetch_slots_bulk<P: Provider<AnyNetwork>>(
653 provider: &P,
654 requests: Vec<(Address, U256)>,
655 block: BlockId,
656 config: BulkCallConfig,
657) -> Vec<(Address, U256, StorageFetchResult<U256>)> {
658 let config = config.normalized();
659 if requests.is_empty() {
660 return Vec::new();
661 }
662 let extractor = config.extractor();
663 let plans = plan_calls(&requests, &config);
664 debug!(
665 slots = requests.len(),
666 calls = plans.len(),
667 dispatch = ?config.dispatch,
668 "bulk storage extraction dispatch"
669 );
670
671 let extractor = &extractor;
672 let call_many_number = match (config.dispatch, block) {
675 (CallDispatch::CallMany, BlockId::Number(number)) => Some(number),
676 _ => None,
677 };
678
679 let Some(number) = call_many_number else {
680 let results: Vec<Vec<_>> = stream::iter(
681 plans
682 .into_iter()
683 .map(|plan| execute_plan(provider, block, plan, extractor)),
684 )
685 .buffer_unordered(config.max_concurrent_calls)
686 .collect()
687 .await;
688 return results.into_iter().flatten().collect();
689 };
690
691 let (conflicting, bundleable): (Vec<_>, Vec<_>) = plans.into_iter().partition(
695 |plan| matches!(plan, CallPlan::Single { target, .. } if *target == MULTICALL3_ADDRESS),
696 );
697 let groups = group_plans_for_call_many(bundleable, config.max_slots_per_request);
698 let group_futs = groups.into_iter().map(|group| async move {
699 match execute_plans_call_many(provider, number, &group, extractor).await {
700 Ok(results) => results,
701 Err(e) => {
702 warn!(
706 error = %e,
707 chunks = group.len(),
708 "eth_callMany dispatch failed; re-dispatching per-call"
709 );
710 let mut results = Vec::new();
711 for plan in group {
712 results.extend(execute_plan(provider, block, plan, extractor).await);
713 }
714 results
715 }
716 }
717 });
718 let mut results: Vec<_> = stream::iter(group_futs)
719 .buffer_unordered(config.max_concurrent_calls)
720 .collect::<Vec<Vec<_>>>()
721 .await
722 .into_iter()
723 .flatten()
724 .collect();
725 for plan in conflicting {
726 results.extend(execute_plan(provider, block, plan, extractor).await);
727 }
728 results
729}
730
731pub fn planned_call_count(requests: &[(Address, U256)], config: &BulkCallConfig) -> usize {
737 plan_calls(requests, &config.normalized()).len()
738}
739
740#[derive(Clone, Debug)]
756pub struct BulkFetcherStatus {
757 consecutive_failures: Arc<std::sync::atomic::AtomicUsize>,
758 latch_threshold: usize,
759}
760
761impl BulkFetcherStatus {
762 pub fn consecutive_override_failures(&self) -> usize {
768 self.consecutive_failures
769 .load(std::sync::atomic::Ordering::Relaxed)
770 }
771
772 pub fn latch_threshold(&self) -> usize {
774 self.latch_threshold
775 }
776
777 pub fn fallback_latched(&self) -> bool {
783 self.consecutive_override_failures() >= self.latch_threshold
784 }
785}
786
787pub fn bulk_call_storage_fetcher<P: Provider<AnyNetwork> + 'static>(
799 provider: Arc<P>,
800 config: BulkCallConfig,
801) -> StorageBatchFetchFn {
802 make_fetcher(provider, config, None).0
803}
804
805pub fn bulk_call_storage_fetcher_with_fallback<P: Provider<AnyNetwork> + 'static>(
815 provider: Arc<P>,
816 config: BulkCallConfig,
817 fallback: StorageBatchFetchFn,
818) -> StorageBatchFetchFn {
819 make_fetcher(provider, config, Some(fallback)).0
820}
821
822pub fn bulk_call_storage_fetcher_with_status<P: Provider<AnyNetwork> + 'static>(
832 provider: Arc<P>,
833 config: BulkCallConfig,
834 fallback: StorageBatchFetchFn,
835) -> (StorageBatchFetchFn, BulkFetcherStatus) {
836 make_fetcher(provider, config, Some(fallback))
837}
838
839fn make_fetcher<P: Provider<AnyNetwork> + 'static>(
840 provider: Arc<P>,
841 config: BulkCallConfig,
842 fallback: Option<StorageBatchFetchFn>,
843) -> (StorageBatchFetchFn, BulkFetcherStatus) {
844 let config = config.normalized();
845 const OVERRIDE_FAILURE_LATCH: usize = 2;
852 let consecutive_failures = Arc::new(std::sync::atomic::AtomicUsize::new(0));
853 let status = BulkFetcherStatus {
854 consecutive_failures: Arc::clone(&consecutive_failures),
855 latch_threshold: OVERRIDE_FAILURE_LATCH,
856 };
857 let fetcher: StorageBatchFetchFn =
858 Arc::new(move |requests: Vec<(Address, U256)>, block: BlockId| {
859 use std::sync::atomic::Ordering;
860 if requests.is_empty() {
861 return Vec::new();
862 }
863 if let Some(fallback) = &fallback
864 && (requests.len() < config.point_read_threshold
865 || consecutive_failures.load(Ordering::Relaxed) >= OVERRIDE_FAILURE_LATCH)
866 {
867 return fallback(requests, block);
868 }
869
870 let bulk_results = match block_in_place_handle() {
877 Ok(handle) => tokio::task::block_in_place(|| {
878 handle.block_on(fetch_slots_bulk(provider.as_ref(), requests, block, config))
879 }),
880 Err(e) => requests
881 .into_iter()
882 .map(|(addr, slot)| (addr, slot, Err(StorageFetchError::Runtime(e.clone()))))
883 .collect(),
884 };
885
886 let Some(fallback) = &fallback else {
887 return bulk_results;
888 };
889
890 if bulk_results.iter().any(|(_, _, r)| r.is_ok()) {
893 consecutive_failures.store(0, Ordering::Relaxed);
894 } else if bulk_results
895 .iter()
896 .any(|(_, _, r)| matches!(r, Err(StorageFetchError::Provider { .. })))
897 {
898 let streak = consecutive_failures.fetch_add(1, Ordering::Relaxed) + 1;
899 if streak == OVERRIDE_FAILURE_LATCH {
900 warn!(
901 streak,
902 "bulk storage extraction failed consecutive batches with provider errors; \
903 latching this fetcher to the point-read fallback (install a fresh fetcher \
904 to retry bulk extraction)"
905 );
906 }
907 }
908
909 let mut repaired = Vec::with_capacity(bulk_results.len());
912 let mut failed: Vec<(Address, U256)> = Vec::new();
913 for (addr, slot, result) in bulk_results {
914 match result {
915 Ok(value) => repaired.push((addr, slot, Ok(value))),
916 Err(_) => failed.push((addr, slot)),
917 }
918 }
919 if !failed.is_empty() {
920 warn!(
921 failed = failed.len(),
922 "bulk storage extraction failed for some slots; repairing via fallback fetcher"
923 );
924 repaired.extend(fallback(failed, block));
925 }
926 repaired
927 });
928 (fetcher, status)
929}
930
931#[derive(Debug, Clone, PartialEq, Eq)]
953pub struct StorageProgram {
954 pub target: Address,
957 pub code: Bytes,
959 pub calldata: Bytes,
961}
962
963pub async fn run_storage_program<P: Provider<AnyNetwork>>(
965 provider: &P,
966 block: BlockId,
967 program: &StorageProgram,
968) -> StorageFetchResult<Bytes> {
969 let mut overrides = StateOverride::default();
970 overrides.insert(
971 program.target,
972 AccountOverride::default().with_code(program.code.clone()),
973 );
974 let tx = TransactionRequest::default()
975 .to(program.target)
976 .input(program.calldata.clone().into());
977 provider
978 .client()
979 .request("eth_call", (tx, block, overrides))
980 .await
981 .map_err(|e| StorageFetchError::provider("eth_call", &e))
982}
983
984pub async fn run_storage_programs<P: Provider<AnyNetwork>>(
991 provider: &P,
992 block: BlockId,
993 programs: &[StorageProgram],
994) -> Vec<StorageFetchResult<Bytes>> {
995 let mut seen = std::collections::HashSet::new();
996 let mut bundle: Vec<usize> = Vec::new();
997 let mut individual: Vec<usize> = Vec::new();
998 for (index, program) in programs.iter().enumerate() {
999 if program.target != MULTICALL3_ADDRESS && seen.insert(program.target) {
1000 bundle.push(index);
1001 } else {
1002 individual.push(index);
1003 }
1004 }
1005 if bundle.len() == 1 {
1007 individual.append(&mut bundle);
1008 }
1009
1010 let mut out: Vec<Option<StorageFetchResult<Bytes>>> = vec![None; programs.len()];
1011
1012 if !bundle.is_empty() {
1013 let mut overrides = StateOverride::default();
1014 overrides.insert(
1015 MULTICALL3_ADDRESS,
1016 AccountOverride::default().with_code(multicall3_runtime_code().clone()),
1017 );
1018 let calls: Vec<IMulticall3::Call3> = bundle
1019 .iter()
1020 .map(|&index| {
1021 let program = &programs[index];
1022 overrides.insert(
1023 program.target,
1024 AccountOverride::default().with_code(program.code.clone()),
1025 );
1026 IMulticall3::Call3 {
1027 target: program.target,
1028 allowFailure: true,
1029 callData: program.calldata.clone(),
1030 }
1031 })
1032 .collect();
1033 let data: Bytes = IMulticall3::aggregate3Call { calls }.abi_encode().into();
1034 let tx = TransactionRequest::default()
1035 .to(MULTICALL3_ADDRESS)
1036 .input(data.into());
1037 let response: Result<Bytes, _> = provider
1038 .client()
1039 .request("eth_call", (tx, block, overrides))
1040 .await;
1041 match response
1042 .map_err(|e| StorageFetchError::provider("eth_call", &e))
1043 .and_then(|bytes| {
1044 IMulticall3::aggregate3Call::abi_decode_returns(&bytes).map_err(|e| {
1045 StorageFetchError::custom(format!("failed to decode aggregate3 response: {e}"))
1046 })
1047 }) {
1048 Ok(results) if results.len() == bundle.len() => {
1049 for (&index, result) in bundle.iter().zip(results) {
1050 out[index] = Some(if result.success {
1051 Ok(result.returnData)
1052 } else {
1053 Err(StorageFetchError::custom(
1054 "storage program subcall failed (allowFailure=true)",
1055 ))
1056 });
1057 }
1058 }
1059 Ok(results) => {
1060 let err = StorageFetchError::custom(format!(
1061 "aggregate3 returned {} results for {} programs",
1062 results.len(),
1063 bundle.len()
1064 ));
1065 for &index in &bundle {
1066 out[index] = Some(Err(err.clone()));
1067 }
1068 }
1069 Err(err) => {
1070 for &index in &bundle {
1071 out[index] = Some(Err(err.clone()));
1072 }
1073 }
1074 }
1075 }
1076
1077 for index in individual {
1078 out[index] = Some(run_storage_program(provider, block, &programs[index]).await);
1079 }
1080
1081 out.into_iter()
1082 .map(|entry| entry.expect("every program resolved"))
1083 .collect()
1084}
1085
1086pub const ACCOUNT_FIELDS_EXTRACTOR_CODE: &[u8] =
1106 &hex!("5f5b803614602057803580318260011b523f8160011b602001526020016001565b3660011b5ff3");
1107
1108#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1111pub struct AccountFieldsSample {
1112 pub balance: U256,
1114 pub code_hash: B256,
1117}
1118
1119pub async fn fetch_account_fields_bulk<P: Provider<AnyNetwork>>(
1128 provider: &P,
1129 addresses: &[Address],
1130 block: BlockId,
1131) -> StorageFetchResult<Vec<(Address, AccountFieldsSample)>> {
1132 if addresses.is_empty() {
1133 return Ok(Vec::new());
1134 }
1135 let mut calldata = Vec::with_capacity(addresses.len() * 32);
1136 for address in addresses {
1137 calldata.extend_from_slice(&[0u8; 12]);
1138 calldata.extend_from_slice(address.as_slice());
1139 }
1140 let program = StorageProgram {
1141 target: MULTICALL3_ADDRESS,
1142 code: Bytes::from_static(ACCOUNT_FIELDS_EXTRACTOR_CODE),
1143 calldata: calldata.into(),
1144 };
1145 let bytes = run_storage_program(provider, block, &program).await?;
1146 if bytes.len() != addresses.len() * 64 {
1147 return Err(StorageFetchError::custom(format!(
1148 "account-fields extractor returned {} bytes, expected {}",
1149 bytes.len(),
1150 addresses.len() * 64
1151 )));
1152 }
1153 Ok(addresses
1154 .iter()
1155 .enumerate()
1156 .map(|(i, address)| {
1157 (
1158 *address,
1159 AccountFieldsSample {
1160 balance: U256::from_be_slice(&bytes[i * 64..i * 64 + 32]),
1161 code_hash: B256::from_slice(&bytes[i * 64 + 32..i * 64 + 64]),
1162 },
1163 )
1164 })
1165 .collect())
1166}
1167
1168pub const BLOCK_CONTEXT_EXTRACTOR_CODE: &[u8] =
1174 &hex!("435f52426020524860405241606052446080524560a0524660c05260e05ff3");
1175
1176#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1179pub struct BlockContextSample {
1180 pub number: u64,
1182 pub timestamp: u64,
1184 pub basefee: U256,
1186 pub coinbase: Address,
1188 pub prevrandao: B256,
1190 pub gas_limit: u64,
1192 pub chain_id: u64,
1194}
1195
1196pub async fn fetch_block_context<P: Provider<AnyNetwork>>(
1199 provider: &P,
1200 block: BlockId,
1201) -> StorageFetchResult<BlockContextSample> {
1202 let program = StorageProgram {
1203 target: MULTICALL3_ADDRESS,
1204 code: Bytes::from_static(BLOCK_CONTEXT_EXTRACTOR_CODE),
1205 calldata: Bytes::new(),
1206 };
1207 let bytes = run_storage_program(provider, block, &program).await?;
1208 if bytes.len() != 7 * 32 {
1209 return Err(StorageFetchError::custom(format!(
1210 "block-context extractor returned {} bytes, expected 224",
1211 bytes.len()
1212 )));
1213 }
1214 let word = |i: usize| U256::from_be_slice(&bytes[i * 32..(i + 1) * 32]);
1215 let to_u64 = |v: U256| u64::try_from(v).unwrap_or(u64::MAX);
1216 Ok(BlockContextSample {
1217 number: to_u64(word(0)),
1218 timestamp: to_u64(word(1)),
1219 basefee: word(2),
1220 coinbase: Address::from_slice(&bytes[3 * 32 + 12..4 * 32]),
1221 prevrandao: B256::from_slice(&bytes[4 * 32..5 * 32]),
1222 gas_limit: to_u64(word(5)),
1223 chain_id: to_u64(word(6)),
1224 })
1225}
1226
1227#[cfg(test)]
1228mod tests {
1229 use super::*;
1230
1231 fn addr(byte: u8) -> Address {
1232 Address::repeat_byte(byte)
1233 }
1234
1235 fn cfg(max_slots: usize, max_targets: usize) -> BulkCallConfig {
1236 BulkCallConfig {
1237 max_slots_per_call: max_slots,
1238 max_targets_per_call: max_targets,
1239 ..BulkCallConfig::default()
1240 }
1241 }
1242
1243 #[test]
1244 fn pack_and_decode_roundtrip() {
1245 let slots = vec![U256::ZERO, U256::from(1u64), U256::MAX];
1246 let packed = pack_slots_calldata(&slots);
1247 assert_eq!(packed.len(), 96);
1248 assert_eq!(&packed[32..64], &U256::from(1u64).to_be_bytes::<32>());
1249 let decoded = decode_packed_values(&packed, 3).expect("exact length");
1250 assert_eq!(decoded, slots);
1251 assert!(decode_packed_values(&packed, 2).is_none());
1252 assert!(decode_packed_values(&packed[..95], 3).is_none());
1253 }
1254
1255 #[test]
1256 fn extractor_constants_are_wellformed() {
1257 assert_eq!(STORAGE_EXTRACTOR_CODE.len(), 23);
1260 assert_eq!(STORAGE_EXTRACTOR_CODE[0], 0x5f, "PUSH0 entry");
1261 assert_eq!(STORAGE_EXTRACTOR_CODE_PRE_SHANGHAI.len(), 25);
1262 assert!(
1263 !STORAGE_EXTRACTOR_CODE_PRE_SHANGHAI.contains(&0x5f),
1264 "pre-Shanghai variant must not use PUSH0"
1265 );
1266 assert!(multicall3_runtime_code().len() > 1_000);
1267 }
1268
1269 #[test]
1270 fn planning_single_small_group() {
1271 let requests = vec![
1272 (addr(0xaa), U256::from(1u64)),
1273 (addr(0xaa), U256::from(2u64)),
1274 ];
1275 let plans = plan_calls(&requests, &cfg(100, 10));
1276 assert_eq!(
1277 plans,
1278 vec![CallPlan::Single {
1279 target: addr(0xaa),
1280 slots: vec![U256::from(1u64), U256::from(2u64)],
1281 }]
1282 );
1283 }
1284
1285 #[test]
1286 fn planning_splits_oversized_target_and_packs_remainder() {
1287 let mut requests: Vec<_> = (0..7u64).map(|i| (addr(0x01), U256::from(i))).collect();
1290 requests.push((addr(0x02), U256::from(99u64)));
1291 let plans = plan_calls(&requests, &cfg(3, 10));
1292 assert_eq!(plans.len(), 3);
1293 assert_eq!(
1294 plans[0],
1295 CallPlan::Single {
1296 target: addr(0x01),
1297 slots: (0..3u64).map(U256::from).collect(),
1298 }
1299 );
1300 assert_eq!(
1301 plans[1],
1302 CallPlan::Single {
1303 target: addr(0x01),
1304 slots: (3..6u64).map(U256::from).collect(),
1305 }
1306 );
1307 assert_eq!(
1308 plans[2],
1309 CallPlan::Multi {
1310 targets: vec![
1311 (addr(0x01), vec![U256::from(6u64)]),
1312 (addr(0x02), vec![U256::from(99u64)]),
1313 ],
1314 }
1315 );
1316 let planned: usize = plans.iter().map(CallPlan::request_slot_count).sum();
1317 assert_eq!(planned, requests.len());
1318 }
1319
1320 #[test]
1321 fn planning_respects_target_budget() {
1322 let requests: Vec<_> = (0..5u8)
1323 .map(|i| (addr(i + 1), U256::from(i as u64)))
1324 .collect();
1325 let plans = plan_calls(&requests, &cfg(100, 2));
1326 assert_eq!(plans.len(), 3);
1328 assert!(matches!(&plans[0], CallPlan::Multi { targets } if targets.len() == 2));
1329 assert!(matches!(&plans[1], CallPlan::Multi { targets } if targets.len() == 2));
1330 assert!(matches!(&plans[2], CallPlan::Single { .. }));
1331 }
1332
1333 #[test]
1334 fn planning_lone_remainder_degrades_to_single_call() {
1335 let requests: Vec<_> = (0..4u64).map(|i| (addr(0x01), U256::from(i))).collect();
1336 let plans = plan_calls(&requests, &cfg(3, 10));
1337 assert_eq!(plans.len(), 2);
1338 assert!(matches!(&plans[1], CallPlan::Single { slots, .. } if slots.len() == 1));
1339 }
1340
1341 #[test]
1342 fn planning_isolates_dispatcher_address_collision() {
1343 let requests = vec![
1344 (MULTICALL3_ADDRESS, U256::from(1u64)),
1345 (addr(0x02), U256::from(2u64)),
1346 (addr(0x03), U256::from(3u64)),
1347 ];
1348 let plans = plan_calls(&requests, &cfg(100, 10));
1349 assert_eq!(
1350 plans[0],
1351 CallPlan::Single {
1352 target: MULTICALL3_ADDRESS,
1353 slots: vec![U256::from(1u64)],
1354 }
1355 );
1356 assert!(matches!(&plans[1], CallPlan::Multi { targets } if targets.len() == 2));
1357 }
1358
1359 #[test]
1360 fn multi_target_overrides_include_dispatcher_and_extractors() {
1361 let plan = CallPlan::Multi {
1362 targets: vec![
1363 (addr(0x02), vec![U256::from(1u64)]),
1364 (addr(0x03), vec![U256::from(2u64)]),
1365 ],
1366 };
1367 let extractor = Bytes::from_static(STORAGE_EXTRACTOR_CODE);
1368 let overrides = overrides_for_plan(&plan, &extractor);
1369 assert_eq!(overrides.len(), 3);
1370 assert_eq!(
1371 overrides[&MULTICALL3_ADDRESS].code.as_ref(),
1372 Some(multicall3_runtime_code())
1373 );
1374 assert_eq!(overrides[&addr(0x02)].code.as_ref(), Some(&extractor));
1375 assert_eq!(overrides[&addr(0x03)].code.as_ref(), Some(&extractor));
1376 }
1377
1378 #[test]
1379 fn call_many_grouping_respects_request_budget() {
1380 let plans = vec![
1381 CallPlan::Single {
1382 target: addr(0x01),
1383 slots: (0..6u64).map(U256::from).collect(),
1384 },
1385 CallPlan::Single {
1386 target: addr(0x02),
1387 slots: (0..6u64).map(U256::from).collect(),
1388 },
1389 CallPlan::Single {
1390 target: addr(0x03),
1391 slots: (0..2u64).map(U256::from).collect(),
1392 },
1393 ];
1394 let groups = group_plans_for_call_many(plans, 10);
1395 assert_eq!(groups.len(), 2);
1397 assert_eq!(groups[0].len(), 1);
1398 assert_eq!(groups[1].len(), 2);
1399 let total: usize = groups
1400 .iter()
1401 .flatten()
1402 .map(CallPlan::request_slot_count)
1403 .sum();
1404 assert_eq!(total, 14);
1405 }
1406
1407 #[test]
1408 fn multi_target_response_decodes_per_target_failures() {
1409 let targets = vec![
1410 (addr(0x02), vec![U256::from(1u64), U256::from(2u64)]),
1411 (addr(0x03), vec![U256::from(3u64)]),
1412 ];
1413 let response = IMulticall3::aggregate3Call::abi_encode_returns(&vec![
1414 IMulticall3::Result {
1415 success: true,
1416 returnData: pack_slots_calldata(&[U256::from(11u64), U256::from(22u64)]),
1417 },
1418 IMulticall3::Result {
1419 success: false,
1420 returnData: Bytes::new(),
1421 },
1422 ]);
1423 let results = decode_multi_target_response(&targets, &response);
1424 assert_eq!(results.len(), 3);
1425 assert!(matches!(results[0], (_, _, Ok(v)) if v == U256::from(11u64)));
1426 assert!(matches!(results[1], (_, _, Ok(v)) if v == U256::from(22u64)));
1427 assert!(results[2].2.is_err());
1428 }
1429}