use std::collections::HashMap;
use std::sync::{Arc, OnceLock};
use alloy_eips::BlockId;
use alloy_primitives::{Address, B256, Bytes, U256, hex};
use alloy_provider::Provider;
use alloy_provider::network::AnyNetwork;
use alloy_rpc_types_eth::TransactionRequest;
use alloy_rpc_types_eth::state::{AccountOverride, StateOverride};
use alloy_sol_types::SolCall;
use futures::stream::{self, StreamExt};
use tracing::{debug, warn};
use crate::cache::{StorageBatchFetchFn, block_in_place_handle};
use crate::errors::{StorageFetchError, StorageFetchResult};
use crate::multicall::{IMulticall3, MULTICALL3_ADDRESS};
pub const STORAGE_EXTRACTOR_CODE: &[u8] = &hex!("5f5b80361460135780355481526020016001565b365ff3");
pub const STORAGE_EXTRACTOR_CODE_PRE_SHANGHAI: &[u8] =
&hex!("60005b80361460145780355481526020016002565b366000f3");
pub fn multicall3_runtime_code() -> &'static Bytes {
static CODE: OnceLock<Bytes> = OnceLock::new();
CODE.get_or_init(|| {
let raw = include_str!("../fixtures/multicall3_runtime.hex");
Bytes::from(hex::decode(raw.trim()).expect("valid multicall3 runtime hex fixture"))
})
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum CallDispatch {
#[default]
PerCall,
CallMany,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct BulkCallConfig {
pub max_slots_per_call: usize,
pub max_targets_per_call: usize,
pub max_concurrent_calls: usize,
pub point_read_threshold: usize,
pub pre_shanghai_extractor: bool,
pub dispatch: CallDispatch,
pub max_slots_per_request: usize,
}
impl Default for BulkCallConfig {
fn default() -> Self {
Self {
max_slots_per_call: 10_000,
max_targets_per_call: 250,
max_concurrent_calls: 4,
point_read_threshold: 2,
pre_shanghai_extractor: false,
dispatch: CallDispatch::PerCall,
max_slots_per_request: 25_000,
}
}
}
impl BulkCallConfig {
fn normalized(self) -> Self {
Self {
max_slots_per_call: self.max_slots_per_call.max(1),
max_targets_per_call: self.max_targets_per_call.max(1),
max_concurrent_calls: self.max_concurrent_calls.max(1),
max_slots_per_request: self.max_slots_per_request.max(1),
..self
}
}
fn extractor(&self) -> Bytes {
if self.pre_shanghai_extractor {
Bytes::from_static(STORAGE_EXTRACTOR_CODE_PRE_SHANGHAI)
} else {
Bytes::from_static(STORAGE_EXTRACTOR_CODE)
}
}
}
pub fn pack_slots_calldata(slots: &[U256]) -> Bytes {
let mut out = Vec::with_capacity(slots.len() * 32);
for slot in slots {
out.extend_from_slice(&slot.to_be_bytes::<32>());
}
out.into()
}
pub fn decode_packed_values(data: &[u8], expected: usize) -> Option<Vec<U256>> {
if data.len() != expected * 32 {
return None;
}
Some(data.chunks_exact(32).map(U256::from_be_slice).collect())
}
pub fn encode_multi_target_calldata(targets: &[(Address, Vec<U256>)]) -> Bytes {
let calls: Vec<IMulticall3::Call3> = targets
.iter()
.map(|(target, slots)| IMulticall3::Call3 {
target: *target,
allowFailure: true,
callData: pack_slots_calldata(slots),
})
.collect();
IMulticall3::aggregate3Call { calls }.abi_encode().into()
}
pub fn decode_multi_target_response(
targets: &[(Address, Vec<U256>)],
response: &[u8],
) -> Vec<(Address, U256, StorageFetchResult<U256>)> {
let decoded = match IMulticall3::aggregate3Call::abi_decode_returns(response) {
Ok(results) if results.len() == targets.len() => results,
Ok(results) => {
return per_target_errors(targets, || {
StorageFetchError::custom(format!(
"aggregate3 returned {} results for {} extraction targets",
results.len(),
targets.len()
))
});
}
Err(e) => {
return per_target_errors(targets, || {
StorageFetchError::custom(format!("failed to decode aggregate3 response: {e}"))
});
}
};
let mut out = Vec::with_capacity(targets.iter().map(|(_, s)| s.len()).sum());
for ((target, slots), result) in targets.iter().zip(decoded) {
if !result.success {
out.extend(slots.iter().map(|slot| {
(
*target,
*slot,
Err(StorageFetchError::custom(
"extractor subcall failed (allowFailure=true); the target may be a precompile",
)),
)
}));
continue;
}
match decode_packed_values(&result.returnData, slots.len()) {
Some(values) => out.extend(
slots
.iter()
.zip(values)
.map(|(slot, value)| (*target, *slot, Ok(value))),
),
None => out.extend(slots.iter().map(|slot| {
(
*target,
*slot,
Err(StorageFetchError::custom(format!(
"extractor at {target} returned {} bytes, expected {}",
result.returnData.len(),
slots.len() * 32
))),
)
})),
}
}
out
}
fn per_target_errors(
targets: &[(Address, Vec<U256>)],
make: impl Fn() -> StorageFetchError,
) -> Vec<(Address, U256, StorageFetchResult<U256>)> {
targets
.iter()
.flat_map(|(target, slots)| slots.iter().map(|slot| (*target, *slot, Err(make()))))
.collect()
}
#[derive(Debug, Clone, PartialEq, Eq)]
enum CallPlan {
Single { target: Address, slots: Vec<U256> },
Multi { targets: Vec<(Address, Vec<U256>)> },
}
impl CallPlan {
fn request_slot_count(&self) -> usize {
match self {
Self::Single { slots, .. } => slots.len(),
Self::Multi { targets } => targets.iter().map(|(_, s)| s.len()).sum(),
}
}
}
fn plan_calls(requests: &[(Address, U256)], config: &BulkCallConfig) -> Vec<CallPlan> {
let mut order: Vec<Address> = Vec::new();
let mut groups: HashMap<Address, Vec<U256>> = HashMap::new();
for (address, slot) in requests {
groups
.entry(*address)
.or_insert_with(|| {
order.push(*address);
Vec::new()
})
.push(*slot);
}
let mut plans = Vec::new();
let mut packable: Vec<(Address, Vec<U256>)> = Vec::new();
for address in order {
let slots = groups.remove(&address).expect("grouped above");
for chunk in slots.chunks(config.max_slots_per_call) {
let full = chunk.len() == config.max_slots_per_call;
if full || address == MULTICALL3_ADDRESS {
plans.push(CallPlan::Single {
target: address,
slots: chunk.to_vec(),
});
} else {
packable.push((address, chunk.to_vec()));
}
}
}
let mut current: Vec<(Address, Vec<U256>)> = Vec::new();
let mut current_slots = 0usize;
let flush =
|current: &mut Vec<(Address, Vec<U256>)>, plans: &mut Vec<CallPlan>| match current.len() {
0 => {}
1 => {
let (target, slots) = current.pop().expect("len checked");
plans.push(CallPlan::Single { target, slots });
}
_ => plans.push(CallPlan::Multi {
targets: std::mem::take(current),
}),
};
for (address, slots) in packable {
let would_overflow = current_slots + slots.len() > config.max_slots_per_call
|| current.len() >= config.max_targets_per_call;
if !current.is_empty() && would_overflow {
flush(&mut current, &mut plans);
current_slots = 0;
}
current_slots += slots.len();
current.push((address, slots));
}
flush(&mut current, &mut plans);
plans
}
fn overrides_for_plan(plan: &CallPlan, extractor: &Bytes) -> StateOverride {
let mut overrides = StateOverride::default();
match plan {
CallPlan::Single { target, .. } => {
overrides.insert(
*target,
AccountOverride::default().with_code(extractor.clone()),
);
}
CallPlan::Multi { targets } => {
overrides.insert(
MULTICALL3_ADDRESS,
AccountOverride::default().with_code(multicall3_runtime_code().clone()),
);
for (target, _) in targets {
overrides.insert(
*target,
AccountOverride::default().with_code(extractor.clone()),
);
}
}
}
overrides
}
fn plan_call_parts(plan: &CallPlan) -> (Address, Bytes) {
match plan {
CallPlan::Single { target, slots } => (*target, pack_slots_calldata(slots)),
CallPlan::Multi { targets } => (MULTICALL3_ADDRESS, encode_multi_target_calldata(targets)),
}
}
fn decode_plan_response(
plan: &CallPlan,
bytes: &[u8],
) -> Vec<(Address, U256, StorageFetchResult<U256>)> {
match plan {
CallPlan::Single { target, slots } => match decode_packed_values(bytes, slots.len()) {
Some(values) => slots
.iter()
.zip(values)
.map(|(slot, value)| (*target, *slot, Ok(value)))
.collect(),
None => slots
.iter()
.map(|slot| {
(
*target,
*slot,
Err(StorageFetchError::custom(format!(
"extractor at {target} returned {} bytes, expected {} — the \
provider may not support eth_call state overrides, or the \
target is a precompile",
bytes.len(),
slots.len() * 32
))),
)
})
.collect(),
},
CallPlan::Multi { targets } => decode_multi_target_response(targets, bytes),
}
}
fn plan_error_results(
plan: &CallPlan,
err: StorageFetchError,
) -> Vec<(Address, U256, StorageFetchResult<U256>)> {
match plan {
CallPlan::Single { target, slots } => slots
.iter()
.map(|slot| (*target, *slot, Err(err.clone())))
.collect(),
CallPlan::Multi { targets } => targets
.iter()
.flat_map(|(target, slots)| {
slots.iter().map({
let err = err.clone();
move |slot| (*target, *slot, Err(err.clone()))
})
})
.collect(),
}
}
async fn execute_plan<P: Provider<AnyNetwork>>(
provider: &P,
block: BlockId,
plan: CallPlan,
extractor: &Bytes,
) -> Vec<(Address, U256, StorageFetchResult<U256>)> {
let overrides = overrides_for_plan(&plan, extractor);
let (to, data) = plan_call_parts(&plan);
let tx = TransactionRequest::default().to(to).input(data.into());
let response: Result<Bytes, _> = provider
.client()
.request("eth_call", (tx, block, overrides))
.await;
match response {
Ok(bytes) => decode_plan_response(&plan, &bytes),
Err(e) => plan_error_results(&plan, StorageFetchError::provider("eth_call", &e)),
}
}
#[derive(Debug, serde::Deserialize)]
struct CallManyEntry {
value: Option<Bytes>,
error: Option<serde_json::Value>,
}
async fn execute_plans_call_many<P: Provider<AnyNetwork>>(
provider: &P,
number: alloy_eips::BlockNumberOrTag,
plans: &[CallPlan],
extractor: &Bytes,
) -> Result<Vec<(Address, U256, StorageFetchResult<U256>)>, StorageFetchError> {
let mut overrides = StateOverride::default();
let mut transactions = Vec::with_capacity(plans.len());
for plan in plans {
for (address, account) in overrides_for_plan(plan, extractor) {
overrides.insert(address, account);
}
let (to, data) = plan_call_parts(plan);
transactions.push(serde_json::json!({ "to": to, "data": data }));
}
let bundles = serde_json::json!([{ "transactions": transactions }]);
let context = serde_json::json!({ "blockNumber": number, "transactionIndex": -1 });
let response: Vec<Vec<CallManyEntry>> = provider
.client()
.request("eth_callMany", (bundles, context, overrides))
.await
.map_err(|e| StorageFetchError::provider("eth_callMany", &e))?;
let entries: Vec<CallManyEntry> = response.into_iter().flatten().collect();
if entries.len() != plans.len() {
return Err(StorageFetchError::custom(format!(
"eth_callMany returned {} results for {} bundled calls",
entries.len(),
plans.len()
)));
}
let mut out = Vec::new();
for (plan, entry) in plans.iter().zip(entries) {
match entry.value {
Some(bytes) => out.extend(decode_plan_response(plan, &bytes)),
None => {
let detail = entry
.error
.map(|e| e.to_string())
.unwrap_or_else(|| "no value returned".to_string());
out.extend(plan_error_results(
plan,
StorageFetchError::custom(format!("eth_callMany transaction failed: {detail}")),
));
}
}
}
Ok(out)
}
fn group_plans_for_call_many(
plans: Vec<CallPlan>,
max_slots_per_request: usize,
) -> Vec<Vec<CallPlan>> {
let mut requests: Vec<Vec<CallPlan>> = Vec::new();
let mut current: Vec<CallPlan> = Vec::new();
let mut current_slots = 0usize;
for plan in plans {
let slots = plan.request_slot_count();
if !current.is_empty() && current_slots + slots > max_slots_per_request {
requests.push(std::mem::take(&mut current));
current_slots = 0;
}
current_slots += slots;
current.push(plan);
}
if !current.is_empty() {
requests.push(current);
}
requests
}
pub async fn fetch_slots_bulk<P: Provider<AnyNetwork>>(
provider: &P,
requests: Vec<(Address, U256)>,
block: BlockId,
config: BulkCallConfig,
) -> Vec<(Address, U256, StorageFetchResult<U256>)> {
let config = config.normalized();
if requests.is_empty() {
return Vec::new();
}
let extractor = config.extractor();
let plans = plan_calls(&requests, &config);
debug!(
slots = requests.len(),
calls = plans.len(),
dispatch = ?config.dispatch,
"bulk storage extraction dispatch"
);
let extractor = &extractor;
let call_many_number = match (config.dispatch, block) {
(CallDispatch::CallMany, BlockId::Number(number)) => Some(number),
_ => None,
};
let Some(number) = call_many_number else {
let results: Vec<Vec<_>> = stream::iter(
plans
.into_iter()
.map(|plan| execute_plan(provider, block, plan, extractor)),
)
.buffer_unordered(config.max_concurrent_calls)
.collect()
.await;
return results.into_iter().flatten().collect();
};
let (conflicting, bundleable): (Vec<_>, Vec<_>) = plans.into_iter().partition(
|plan| matches!(plan, CallPlan::Single { target, .. } if *target == MULTICALL3_ADDRESS),
);
let groups = group_plans_for_call_many(bundleable, config.max_slots_per_request);
let group_futs = groups.into_iter().map(|group| async move {
match execute_plans_call_many(provider, number, &group, extractor).await {
Ok(results) => results,
Err(e) => {
warn!(
error = %e,
chunks = group.len(),
"eth_callMany dispatch failed; re-dispatching per-call"
);
let mut results = Vec::new();
for plan in group {
results.extend(execute_plan(provider, block, plan, extractor).await);
}
results
}
}
});
let mut results: Vec<_> = stream::iter(group_futs)
.buffer_unordered(config.max_concurrent_calls)
.collect::<Vec<Vec<_>>>()
.await
.into_iter()
.flatten()
.collect();
for plan in conflicting {
results.extend(execute_plan(provider, block, plan, extractor).await);
}
results
}
pub fn planned_call_count(requests: &[(Address, U256)], config: &BulkCallConfig) -> usize {
plan_calls(requests, &config.normalized()).len()
}
#[derive(Clone, Debug)]
pub struct BulkFetcherStatus {
consecutive_failures: Arc<std::sync::atomic::AtomicUsize>,
latch_threshold: usize,
}
impl BulkFetcherStatus {
pub fn consecutive_override_failures(&self) -> usize {
self.consecutive_failures
.load(std::sync::atomic::Ordering::Relaxed)
}
pub fn latch_threshold(&self) -> usize {
self.latch_threshold
}
pub fn fallback_latched(&self) -> bool {
self.consecutive_override_failures() >= self.latch_threshold
}
}
pub fn bulk_call_storage_fetcher<P: Provider<AnyNetwork> + 'static>(
provider: Arc<P>,
config: BulkCallConfig,
) -> StorageBatchFetchFn {
make_fetcher(provider, config, None).0
}
pub fn bulk_call_storage_fetcher_with_fallback<P: Provider<AnyNetwork> + 'static>(
provider: Arc<P>,
config: BulkCallConfig,
fallback: StorageBatchFetchFn,
) -> StorageBatchFetchFn {
make_fetcher(provider, config, Some(fallback)).0
}
pub fn bulk_call_storage_fetcher_with_status<P: Provider<AnyNetwork> + 'static>(
provider: Arc<P>,
config: BulkCallConfig,
fallback: StorageBatchFetchFn,
) -> (StorageBatchFetchFn, BulkFetcherStatus) {
make_fetcher(provider, config, Some(fallback))
}
fn make_fetcher<P: Provider<AnyNetwork> + 'static>(
provider: Arc<P>,
config: BulkCallConfig,
fallback: Option<StorageBatchFetchFn>,
) -> (StorageBatchFetchFn, BulkFetcherStatus) {
let config = config.normalized();
const OVERRIDE_FAILURE_LATCH: usize = 2;
let consecutive_failures = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let status = BulkFetcherStatus {
consecutive_failures: Arc::clone(&consecutive_failures),
latch_threshold: OVERRIDE_FAILURE_LATCH,
};
let fetcher: StorageBatchFetchFn =
Arc::new(move |requests: Vec<(Address, U256)>, block: BlockId| {
use std::sync::atomic::Ordering;
if requests.is_empty() {
return Vec::new();
}
if let Some(fallback) = &fallback
&& (requests.len() < config.point_read_threshold
|| consecutive_failures.load(Ordering::Relaxed) >= OVERRIDE_FAILURE_LATCH)
{
return fallback(requests, block);
}
let bulk_results = match block_in_place_handle() {
Ok(handle) => tokio::task::block_in_place(|| {
handle.block_on(fetch_slots_bulk(provider.as_ref(), requests, block, config))
}),
Err(e) => requests
.into_iter()
.map(|(addr, slot)| (addr, slot, Err(StorageFetchError::Runtime(e.clone()))))
.collect(),
};
let Some(fallback) = &fallback else {
return bulk_results;
};
if bulk_results.iter().any(|(_, _, r)| r.is_ok()) {
consecutive_failures.store(0, Ordering::Relaxed);
} else if bulk_results
.iter()
.any(|(_, _, r)| matches!(r, Err(StorageFetchError::Provider { .. })))
{
let streak = consecutive_failures.fetch_add(1, Ordering::Relaxed) + 1;
if streak == OVERRIDE_FAILURE_LATCH {
warn!(
streak,
"bulk storage extraction failed consecutive batches with provider errors; \
latching this fetcher to the point-read fallback (install a fresh fetcher \
to retry bulk extraction)"
);
}
}
let mut repaired = Vec::with_capacity(bulk_results.len());
let mut failed: Vec<(Address, U256)> = Vec::new();
for (addr, slot, result) in bulk_results {
match result {
Ok(value) => repaired.push((addr, slot, Ok(value))),
Err(_) => failed.push((addr, slot)),
}
}
if !failed.is_empty() {
warn!(
failed = failed.len(),
"bulk storage extraction failed for some slots; repairing via fallback fetcher"
);
repaired.extend(fallback(failed, block));
}
repaired
});
(fetcher, status)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StorageProgram {
pub target: Address,
pub code: Bytes,
pub calldata: Bytes,
}
pub async fn run_storage_program<P: Provider<AnyNetwork>>(
provider: &P,
block: BlockId,
program: &StorageProgram,
) -> StorageFetchResult<Bytes> {
let mut overrides = StateOverride::default();
overrides.insert(
program.target,
AccountOverride::default().with_code(program.code.clone()),
);
let tx = TransactionRequest::default()
.to(program.target)
.input(program.calldata.clone().into());
provider
.client()
.request("eth_call", (tx, block, overrides))
.await
.map_err(|e| StorageFetchError::provider("eth_call", &e))
}
pub async fn run_storage_programs<P: Provider<AnyNetwork>>(
provider: &P,
block: BlockId,
programs: &[StorageProgram],
) -> Vec<StorageFetchResult<Bytes>> {
let mut seen = std::collections::HashSet::new();
let mut bundle: Vec<usize> = Vec::new();
let mut individual: Vec<usize> = Vec::new();
for (index, program) in programs.iter().enumerate() {
if program.target != MULTICALL3_ADDRESS && seen.insert(program.target) {
bundle.push(index);
} else {
individual.push(index);
}
}
if bundle.len() == 1 {
individual.append(&mut bundle);
}
let mut out: Vec<Option<StorageFetchResult<Bytes>>> = vec![None; programs.len()];
if !bundle.is_empty() {
let mut overrides = StateOverride::default();
overrides.insert(
MULTICALL3_ADDRESS,
AccountOverride::default().with_code(multicall3_runtime_code().clone()),
);
let calls: Vec<IMulticall3::Call3> = bundle
.iter()
.map(|&index| {
let program = &programs[index];
overrides.insert(
program.target,
AccountOverride::default().with_code(program.code.clone()),
);
IMulticall3::Call3 {
target: program.target,
allowFailure: true,
callData: program.calldata.clone(),
}
})
.collect();
let data: Bytes = IMulticall3::aggregate3Call { calls }.abi_encode().into();
let tx = TransactionRequest::default()
.to(MULTICALL3_ADDRESS)
.input(data.into());
let response: Result<Bytes, _> = provider
.client()
.request("eth_call", (tx, block, overrides))
.await;
match response
.map_err(|e| StorageFetchError::provider("eth_call", &e))
.and_then(|bytes| {
IMulticall3::aggregate3Call::abi_decode_returns(&bytes).map_err(|e| {
StorageFetchError::custom(format!("failed to decode aggregate3 response: {e}"))
})
}) {
Ok(results) if results.len() == bundle.len() => {
for (&index, result) in bundle.iter().zip(results) {
out[index] = Some(if result.success {
Ok(result.returnData)
} else {
Err(StorageFetchError::custom(
"storage program subcall failed (allowFailure=true)",
))
});
}
}
Ok(results) => {
let err = StorageFetchError::custom(format!(
"aggregate3 returned {} results for {} programs",
results.len(),
bundle.len()
));
for &index in &bundle {
out[index] = Some(Err(err.clone()));
}
}
Err(err) => {
for &index in &bundle {
out[index] = Some(Err(err.clone()));
}
}
}
}
for index in individual {
out[index] = Some(run_storage_program(provider, block, &programs[index]).await);
}
out.into_iter()
.map(|entry| entry.expect("every program resolved"))
.collect()
}
pub const ACCOUNT_FIELDS_EXTRACTOR_CODE: &[u8] =
&hex!("5f5b803614602057803580318260011b523f8160011b602001526020016001565b3660011b5ff3");
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct AccountFieldsSample {
pub balance: U256,
pub code_hash: B256,
}
pub async fn fetch_account_fields_bulk<P: Provider<AnyNetwork>>(
provider: &P,
addresses: &[Address],
block: BlockId,
) -> StorageFetchResult<Vec<(Address, AccountFieldsSample)>> {
if addresses.is_empty() {
return Ok(Vec::new());
}
let mut calldata = Vec::with_capacity(addresses.len() * 32);
for address in addresses {
calldata.extend_from_slice(&[0u8; 12]);
calldata.extend_from_slice(address.as_slice());
}
let program = StorageProgram {
target: MULTICALL3_ADDRESS,
code: Bytes::from_static(ACCOUNT_FIELDS_EXTRACTOR_CODE),
calldata: calldata.into(),
};
let bytes = run_storage_program(provider, block, &program).await?;
if bytes.len() != addresses.len() * 64 {
return Err(StorageFetchError::custom(format!(
"account-fields extractor returned {} bytes, expected {}",
bytes.len(),
addresses.len() * 64
)));
}
Ok(addresses
.iter()
.enumerate()
.map(|(i, address)| {
(
*address,
AccountFieldsSample {
balance: U256::from_be_slice(&bytes[i * 64..i * 64 + 32]),
code_hash: B256::from_slice(&bytes[i * 64 + 32..i * 64 + 64]),
},
)
})
.collect())
}
pub const BLOCK_CONTEXT_EXTRACTOR_CODE: &[u8] =
&hex!("435f52426020524860405241606052446080524560a0524660c05260e05ff3");
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct BlockContextSample {
pub number: u64,
pub timestamp: u64,
pub basefee: U256,
pub coinbase: Address,
pub prevrandao: B256,
pub gas_limit: u64,
pub chain_id: u64,
}
pub async fn fetch_block_context<P: Provider<AnyNetwork>>(
provider: &P,
block: BlockId,
) -> StorageFetchResult<BlockContextSample> {
let program = StorageProgram {
target: MULTICALL3_ADDRESS,
code: Bytes::from_static(BLOCK_CONTEXT_EXTRACTOR_CODE),
calldata: Bytes::new(),
};
let bytes = run_storage_program(provider, block, &program).await?;
if bytes.len() != 7 * 32 {
return Err(StorageFetchError::custom(format!(
"block-context extractor returned {} bytes, expected 224",
bytes.len()
)));
}
let word = |i: usize| U256::from_be_slice(&bytes[i * 32..(i + 1) * 32]);
let to_u64 = |v: U256| u64::try_from(v).unwrap_or(u64::MAX);
Ok(BlockContextSample {
number: to_u64(word(0)),
timestamp: to_u64(word(1)),
basefee: word(2),
coinbase: Address::from_slice(&bytes[3 * 32 + 12..4 * 32]),
prevrandao: B256::from_slice(&bytes[4 * 32..5 * 32]),
gas_limit: to_u64(word(5)),
chain_id: to_u64(word(6)),
})
}
#[cfg(test)]
mod tests {
use super::*;
fn addr(byte: u8) -> Address {
Address::repeat_byte(byte)
}
fn cfg(max_slots: usize, max_targets: usize) -> BulkCallConfig {
BulkCallConfig {
max_slots_per_call: max_slots,
max_targets_per_call: max_targets,
..BulkCallConfig::default()
}
}
#[test]
fn pack_and_decode_roundtrip() {
let slots = vec![U256::ZERO, U256::from(1u64), U256::MAX];
let packed = pack_slots_calldata(&slots);
assert_eq!(packed.len(), 96);
assert_eq!(&packed[32..64], &U256::from(1u64).to_be_bytes::<32>());
let decoded = decode_packed_values(&packed, 3).expect("exact length");
assert_eq!(decoded, slots);
assert!(decode_packed_values(&packed, 2).is_none());
assert!(decode_packed_values(&packed[..95], 3).is_none());
}
#[test]
fn extractor_constants_are_wellformed() {
assert_eq!(STORAGE_EXTRACTOR_CODE.len(), 23);
assert_eq!(STORAGE_EXTRACTOR_CODE[0], 0x5f, "PUSH0 entry");
assert_eq!(STORAGE_EXTRACTOR_CODE_PRE_SHANGHAI.len(), 25);
assert!(
!STORAGE_EXTRACTOR_CODE_PRE_SHANGHAI.contains(&0x5f),
"pre-Shanghai variant must not use PUSH0"
);
assert!(multicall3_runtime_code().len() > 1_000);
}
#[test]
fn planning_single_small_group() {
let requests = vec![
(addr(0xaa), U256::from(1u64)),
(addr(0xaa), U256::from(2u64)),
];
let plans = plan_calls(&requests, &cfg(100, 10));
assert_eq!(
plans,
vec![CallPlan::Single {
target: addr(0xaa),
slots: vec![U256::from(1u64), U256::from(2u64)],
}]
);
}
#[test]
fn planning_splits_oversized_target_and_packs_remainder() {
let mut requests: Vec<_> = (0..7u64).map(|i| (addr(0x01), U256::from(i))).collect();
requests.push((addr(0x02), U256::from(99u64)));
let plans = plan_calls(&requests, &cfg(3, 10));
assert_eq!(plans.len(), 3);
assert_eq!(
plans[0],
CallPlan::Single {
target: addr(0x01),
slots: (0..3u64).map(U256::from).collect(),
}
);
assert_eq!(
plans[1],
CallPlan::Single {
target: addr(0x01),
slots: (3..6u64).map(U256::from).collect(),
}
);
assert_eq!(
plans[2],
CallPlan::Multi {
targets: vec![
(addr(0x01), vec![U256::from(6u64)]),
(addr(0x02), vec![U256::from(99u64)]),
],
}
);
let planned: usize = plans.iter().map(CallPlan::request_slot_count).sum();
assert_eq!(planned, requests.len());
}
#[test]
fn planning_respects_target_budget() {
let requests: Vec<_> = (0..5u8)
.map(|i| (addr(i + 1), U256::from(i as u64)))
.collect();
let plans = plan_calls(&requests, &cfg(100, 2));
assert_eq!(plans.len(), 3);
assert!(matches!(&plans[0], CallPlan::Multi { targets } if targets.len() == 2));
assert!(matches!(&plans[1], CallPlan::Multi { targets } if targets.len() == 2));
assert!(matches!(&plans[2], CallPlan::Single { .. }));
}
#[test]
fn planning_lone_remainder_degrades_to_single_call() {
let requests: Vec<_> = (0..4u64).map(|i| (addr(0x01), U256::from(i))).collect();
let plans = plan_calls(&requests, &cfg(3, 10));
assert_eq!(plans.len(), 2);
assert!(matches!(&plans[1], CallPlan::Single { slots, .. } if slots.len() == 1));
}
#[test]
fn planning_isolates_dispatcher_address_collision() {
let requests = vec![
(MULTICALL3_ADDRESS, U256::from(1u64)),
(addr(0x02), U256::from(2u64)),
(addr(0x03), U256::from(3u64)),
];
let plans = plan_calls(&requests, &cfg(100, 10));
assert_eq!(
plans[0],
CallPlan::Single {
target: MULTICALL3_ADDRESS,
slots: vec![U256::from(1u64)],
}
);
assert!(matches!(&plans[1], CallPlan::Multi { targets } if targets.len() == 2));
}
#[test]
fn multi_target_overrides_include_dispatcher_and_extractors() {
let plan = CallPlan::Multi {
targets: vec![
(addr(0x02), vec![U256::from(1u64)]),
(addr(0x03), vec![U256::from(2u64)]),
],
};
let extractor = Bytes::from_static(STORAGE_EXTRACTOR_CODE);
let overrides = overrides_for_plan(&plan, &extractor);
assert_eq!(overrides.len(), 3);
assert_eq!(
overrides[&MULTICALL3_ADDRESS].code.as_ref(),
Some(multicall3_runtime_code())
);
assert_eq!(overrides[&addr(0x02)].code.as_ref(), Some(&extractor));
assert_eq!(overrides[&addr(0x03)].code.as_ref(), Some(&extractor));
}
#[test]
fn call_many_grouping_respects_request_budget() {
let plans = vec![
CallPlan::Single {
target: addr(0x01),
slots: (0..6u64).map(U256::from).collect(),
},
CallPlan::Single {
target: addr(0x02),
slots: (0..6u64).map(U256::from).collect(),
},
CallPlan::Single {
target: addr(0x03),
slots: (0..2u64).map(U256::from).collect(),
},
];
let groups = group_plans_for_call_many(plans, 10);
assert_eq!(groups.len(), 2);
assert_eq!(groups[0].len(), 1);
assert_eq!(groups[1].len(), 2);
let total: usize = groups
.iter()
.flatten()
.map(CallPlan::request_slot_count)
.sum();
assert_eq!(total, 14);
}
#[test]
fn multi_target_response_decodes_per_target_failures() {
let targets = vec![
(addr(0x02), vec![U256::from(1u64), U256::from(2u64)]),
(addr(0x03), vec![U256::from(3u64)]),
];
let response = IMulticall3::aggregate3Call::abi_encode_returns(&vec![
IMulticall3::Result {
success: true,
returnData: pack_slots_calldata(&[U256::from(11u64), U256::from(22u64)]),
},
IMulticall3::Result {
success: false,
returnData: Bytes::new(),
},
]);
let results = decode_multi_target_response(&targets, &response);
assert_eq!(results.len(), 3);
assert!(matches!(results[0], (_, _, Ok(v)) if v == U256::from(11u64)));
assert!(matches!(results[1], (_, _, Ok(v)) if v == U256::from(22u64)));
assert!(results[2].2.is_err());
}
}