let logs = provider.get_logs(&filter).await?;<\/p>\n
While using Alloy, I kept wondering what a single line like the one above actually does. So I read the relevant parts of Alloy and Reth and wrote down what I\u00a0found.<\/p>\n
We follow one call, eth_getLogs, over HTTP, from the typed Filter in your process through Reth\u2019s log lookup and back to a Vec<Log>. The Alloy side of this path is shared by many ordinary request and response RPC calls, while the log search inside Reth is specific to eth_getLogs. Along the way, we see where the JSON is created, when the request is sent, and how the node finds matching\u00a0logs.<\/p>\n
Reading Rust and having used Alloy before is enough to follow along. We trace Alloy\u2019s default provider implementation down to individual poll calls, then follow the main range query path inside Reth. The code is pinned to Alloy 2.1.1 at <\/a>ffaf1da7<\/a> and Reth 2.4.1 at <\/a>f969bd4d<\/a>. In some code blocks, unrelated parts have been omitted to keep the focus on the path being discussed.<\/p>\n Let\u2019s start with an overview of what is happening in the process, so we can dive deeper later and understand the details. An Ethereum RPC call is, at its core, sending one JSON-RPC 2.0 message and receiving one in response. The node is a separate program. You can\u2019t call its functions directly; you can only send it a message and wait for a reply. Alloy\u2019s whole job on your side is to hide that: it takes your typed Rust values, turns them into that JSON, sends it, waits, and turns the node\u2019s answer back into Rust types so you get a clean Vec<Log> instead of a blob of\u00a0text.<\/p>\n The process overview is roughly\u00a0this:<\/p>\n Alloy serializes:<\/strong> your typed call becomes the request\u00a0JSON.Reth deserializes, processes, serializes:<\/strong> the request JSON becomes a Filter, the filter finds its logs, the logs become the response\u00a0JSON.Alloy deserializes:<\/strong> the response JSON becomes your Vec<Log>.<\/p>\n Here is the call we are going to follow, with the request details filled\u00a0in:<\/p>\n let filter = Filter::new() let logs = provider.get_logs(&filter).await?;<\/p>\n Alloy serializes that filter into JSON-RPC parameters so the server can understand what we need. The request looks like\u00a0this:<\/p>\n { Reth answers in the same JSON-RPC shape, including the same id and carrying either a result or an error. A successful response contains an array of\u00a0logs:<\/p>\n { JSONs are the only thing that travel between the two programs. Everything else is private to one side or the\u00a0other.<\/p>\n Before following the code, let\u2019s open the filter itself, because it is the payload of the whole request. In Alloy, a Filter stores three\u00a0fields:<\/p>\n \/\/ crates\/rpc-types-eth\/src\/filter.rs The block_option says which blocks to search. It can be a range (fromBlock\/toBlock) or a specific blockHash. The address field says which contract address, or addresses, the logs must come from; if it is empty, logs from any address match. The topics array contains up to four topic filters; an empty slot matches anything. They are positional, so a filter can be valid while targeting the wrong topic\u00a0slot.<\/p>\n Ethereum logs have up to four topics. For non-anonymous Solidity events, topics[0] is the hash of the event signature. For ERC-20 transfers, that is keccak256(“Transfer(address,address,uint256)”) -the 0xddf252ad… value in the request above. The remaining slots correspond to the indexed event arguments, in\u00a0order.<\/p>\n event Transfer(address indexed from, address indexed to, uint256 value);<\/p>\n The layout\u00a0is:<\/p>\n topics[0] = keccak256(“Transfer(address,address,uint256)”) value is not indexed, so it is not a topic. It lives in the log data, which means you cannot filter on it with eth_getLogs.<\/p>\n The topic syntax is positional:<\/p>\n { This means: topic0 must be the Transfer signature, topic1 can be anything, topic2 must be the given address, topic3 can be anything. Get a position wrong, and the query still succeeds. It just quietly matches nothing or the wrong\u00a0thing.<\/p>\n Now we can look at what Alloy\u00a0does.<\/p>\n Remember what we are calling the function on: Provider. But what exactly is\u00a0that?<\/p>\n provider.get_logs(&filter).await?;<\/p>\n It looks like a single function call, but several different jobs have to happen before the request reaches the node. Someone needs to provide the API you call, reuse the connection to the node, turn your function call into a JSON-RPC request, and send that request over the\u00a0network.<\/p>\n Instead of putting all of these responsibilities into one place, Alloy splits them across four roles: Provider, RootProvider, RpcClient, and Transport:<\/p>\n The Provider is the object your application interacts with. Whenever you call methods like get_logs, get_balance, or send_transaction, you’re calling methods on a Provider.<\/p>\n get_logs has a default implementation on the Provider trait: it obtains the shared RpcClient from the RootProvider and uses it to perform the\u00a0request.<\/p>\n \/\/ crates\/provider\/src\/provider\/trait.rs \/\/ …<\/p>\n \/\/\/ Returns the RPC client used to send requests. \/\/ …<\/p>\n \/\/\/ Retrieves a [`Vec<Log>`] with the given [`Filter`]. One scope note before we continue: this article follows this default implementation. A provider may override get_logs entirely.<\/p>\n The RootProvider sits at the base of every provider stack, and for our call it does exactly one thing: it holds the shared RpcClient and hands out references to\u00a0it.<\/p>\n \/\/ crates\/provider\/src\/provider\/root.rs pub(crate) struct RootProviderInner<N: Network = Ethereum> { Three fields, and only the first matters to our call. client is the shared RpcClient, the next layer down. heart serves features like watching new blocks and waiting for transactions to be mined; a one-shot read like get_logs never touches it. And _network is a zero-sized PhantomData marker: it holds no data; it only pins the stack to one network’s types at compile\u00a0time.<\/p>\n Providers point back at the same root, so they all share one client and, below it, one connection pool.<\/p>\n The RpcClient converts your Rust function call into a JSON-RPC request: it builds the request, assigns an id, hands the finished request to the transport, and converts the reply back into Rust\u00a0types.<\/p>\n Notice what it does not<\/em> know: the RpcClient knows nothing about Ethereum. Its only job is to speak JSON-RPC: build a request, attach an id, read a response. It holds the transport it sends through and the counter it draws ids from. (The full type also stores is_local andpoll_interval, plus a pubsub handle when that feature is enabled -none of which our call touches.)<\/p>\n \/\/ crates\/rpc-client\/src\/client.rs A note on that id: over HTTP, the reply is already tied to the request by the request\/response exchange itself -one POST, one response- and Alloy never even compares it; we\u2019ll watch it get discarded on the return trip. The id becomes load-bearing on batches and on multiplexed transports like WebSocket, where many requests share one connection and responses can arrive out of order; there, the id is what routes each response back to its\u00a0caller.<\/p>\n The Transport is the only layer that actually talks to the network. Its job: deliver an already-serialized JSON-RPC packet to the node and hand back the reply. It doesn’t know what eth_getLogs means, and it doesn’t know Ethereum. It moves\u00a0packets.<\/p>\n \/\/ crates\/transport\/src\/boxed.rs That is the whole reason it\u2019s four roles and not one. The Provider speaks Ethereum, the RpcClient speaks only JSON-RPC, the RootProvider sits between them holding the client they all share, and the Transport speaks only packets. Each role strips away one concern, and the call simply descends from the one that knows the most to the one that knows the\u00a0least.<\/p>\n Now let\u2019s follow the call\u00a0down.<\/p>\n This is the overview of the function call sequence, and we will look at it in detail by reading the\u00a0code:<\/p>\n Let\u2019s return to the function implementation we\u00a0call:<\/p>\n \/\/\/ Retrieves a [`Vec<Log>`] with the given [`Filter`]. self.client() borrows the shared client from the RootProvider -a ClientRef, which is simply a reference to the RpcClientInner we just\u00a0saw.<\/p>\n request is the RpcClientInner \u2019s method, and does not send anything. It builds the request and wraps it in an RpcCall -a\u00a0future:<\/p>\n \/\/ crates\/rpc-client\/src\/client.rs \/\/ …<\/p>\n pub fn make_request<Params: RpcSend>( At this point, not a single byte has gone to the network. make_request produced a Request<Params>: the method name, the id reserved for this call, and the params, which are still the Rust values you built. That id is the 0 you saw in the request JSON above. Serialization has not happened\u00a0yet.<\/p>\n After RpcCall::new, an RpcCall<Params, Vec<Log>>: a future holding the request, the transport, and the response type it will eventually deserialize into.<\/p>\n When you\u00a0.await the RpcCall, two poll functions come into play: an outer one on the RpcCall itself, and an inner one on its state machine. The outer drives the inner and then deserializes; the inner sends the request and waits for the\u00a0reply.<\/p>\n The outer poll, in\u00a0full:<\/p>\n \/\/ crates\/rpc-client\/src\/call.rs It is short because it does only two things. It polls the inner state machine (this.state) -ready! returns Pending early if the state isn’t finished- and once the state returns the raw response, try_deserialize_ok turns that raw JSON into the typed Vec<Log>; map then applies an optional conversion, the identity for get_logs. We’ll come back to the deserialize step on the return trip. First, the machine underneath.<\/p>\n The inner poll is the state machine that sends the request and waits for the reply. In\u00a0full:<\/p>\n \/\/ crates\/rpc-client\/src\/call.rs let request = request.take().expect(“no request”); The machine has two points where it can pause: is the transport ready, and has the node replied? Over HTTP, only the second one ever actually pauses. Everything else is serialization and bookkeeping. Let\u2019s take the arms one by\u00a0one.<\/p>\n The state machine starts here, and the arm does three things in order: ask the transport if it\u2019s ready, serialize the request, and hand it\u00a0over.<\/p>\n First, the readiness check:<\/p>\n \/\/ crates\/rpc-client\/src\/call.rs For an HTTP transport, this always returns\u00a0Ready:<\/p>\n \/\/ crates\/transport-http\/src\/reqwest_transport.rs Second, serialization: where the request becomes\u00a0JSON.<\/p>\n \/\/ crates\/rpc-client\/src\/call.rs – logging trimmed What does serialize actually do? It\u2019s\u00a0short:<\/p>\n \/\/ crates\/json-rpc\/src\/request.rs to_raw_value(&self) hands the request to serde, which dispatches back to Request \u2018s own Serialize impl. That\u2019s where the envelope from the overview is assembled, field by\u00a0field:<\/p>\n \/\/ crates\/json-rpc\/src\/request.rs let mut map = serializer.serialize_map(Some(3 + sized_params as usize))?; \/\/ Params may be omitted if it is 0-sized map.serialize_entry(“id”, &self.meta.id)?; There they are: method, params, id, jsonrpc: “2.0”; the four fields of the request JSON from the overview, written out one entry at a time. Serializing params recursively serializes your Filter, which is how the typed addresses and topics become the hex strings on the\u00a0wire.<\/p>\n Third, the handoff. connection.call(req.into()) passes the serialized packet to the transport and returns a future. The\u00a0.into() call wraps our single request in RequestPacket::Single. RequestPacket is the wire unit that can carry either a single request or a batch. But where does the returned future come\u00a0from?<\/p>\n \/\/ crates\/transport-http\/src\/reqwest_transport.rs \/\/ poll_ready shown above<\/p>\n #[inline] self.clone() copies the transport handle so the future can own it (do_reqwest takes self by value and the clone is cheap, because a reqwest::Client is itself a handle to a shared connection pool). Then this.do_reqwest(req) constructs<\/em> its future, and Box::pin wraps it into the type the state machine stores. The span\/instrument pair is observability, not control flow: it tags every log line the future emits with the transport’s URL. None of it changes how the request is\u00a0sent.<\/p>\n What does not happen in call is any networking: async functions in Rust are lazy, so at this point not a single byte has moved. The boxed future is the fut the arm stores as it moves\u00a0on:<\/p>\n self.set(Self::AwaitingResponse { fut });<\/p>\n The POST only begins when the next arm polls it for the first\u00a0time.<\/p>\n \/\/ crates\/rpc-client\/src\/call.rs This is the second wait, and the real one: fut.poll(cx) is where the await truly waits on the network. The future stays Pending, the task yields so the runtime can run other work, and nothing happens until the node’s reply comes back. Most of that time belongs to the network and to\u00a0Reth.<\/p>\n That first fut.poll(cx) is also what finally runs do_reqwest, a thin wrapper over\u00a0reqwest:<\/p>\n \/\/ crates\/transport-http\/src\/reqwest_transport.rs – logging trimmed \/\/ Unpack data from the response body. We do this regardless of if !status.is_success() { return Err(TransportErrorKind::http_error( \/\/ Deserialize a Box<RawValue> from the body. If deserialization fails, return .json(&req) writes the JSON body,\u00a0.post(url).send() performs the HTTP POST, resp.bytes() reads the reply. This is where the request finally leaves the process. While the inner poll sits Pending here, the bytes are on the network, and control has passed to\u00a0Reth.<\/p>\n When the reply does arrive, the arm checks its shape. We sent a single request, so a batch reply would be a protocol violation; that\u2019s the Batch arm turning into an error. A transport failure, such as a refused connection or a timeout, surfaces as the Err arm. The normal case, Single, goes into transform_response, which turns the already-parsed reply into a Rust Result. This is the return trip. We\u2019ll walk through its code after Reth has done its\u00a0part.<\/p>\n CallStateProj::Complete => { It\u2019s a contract, not logic: a Rust future should not be polled again after it has returned Ready.\u00a0.await respects this automatically.<\/p>\n While Alloy waits in AwaitingResponse, the request is inside the node. This is the busiest of the three movements: Reth deserializes the request, does the actual work, and serializes the reply. We followed Alloy down to individual polls; here we zoom out one level and follow the main path\u00a0only.<\/p>\n Here is the whole path in\u00a0outline:<\/p>\n Route.<\/strong> jsonrpsee parses the request and rebuilds the typed Filter, then calls the handler for eth_getLogs.Resolve.<\/strong> A filter names either one block hash or a range. A range may use tags like latest, so it becomes two concrete numbers that are checked against the chain head, pruned history, and the configured range\u00a0cap.Search.<\/strong> Cheap pass first: block headers carry a bloom filter (we will see what it is) that answers \u201ccould this block match?\u201d. Only the blocks that say yes get their receipts opened -from cache if recent, from disk if not- and only there does the exact match\u00a0run.Return.<\/strong> Matching logs are stamped with their block and transaction context and serialized back into the response\u00a0JSON.<\/p>\n The HTTP request first reaches jsonrpsee; it is the JSON-RPC library Reth uses. It parses the JSON-RPC request and uses the method field to decide which handler receives the request. That mapping is declared by a\u00a0trait:<\/p>\n \/\/ reth: crates\/rpc\/rpc-eth-api\/src\/filter.rs \/\/\/ Returns logs matching given filter object. The namespace eth and method name getLogs form eth_getLogs. The signature tells jsonrpsee what to do with the parameters: deserialize the request’s filter object into a Rust Filter, then call\u00a0logs.<\/p>\n Reth imports that Filter, along with the Log type it will return, from alloy-rpc-types-eth -the same RPC type definitions Alloy uses on the client side. The two programs hold separate Rust values, but they agree on the JSON representation, so the filter Alloy serialized is reconstructed field by field inside the\u00a0node.<\/p>\n With the conversion done, jsonrpsee calls Reth’s\u00a0handler:<\/p>\n \/\/ reth: crates\/rpc\/rpc\/src\/eth\/filter.rs \/\/\/ Handler for `eth_getLogs` From here on, Reth\u2019s code works with typed Rust values rather than JSON. The block selectors, the USDC address, and the Transfer topic are typed fields inside Filter\u00a0. The handler adds the node\u2019s configured query limits and forwards the request through one small\u00a0wrapper:<\/p>\n async fn logs_for_filter( A filter can identify one block by hash or describe a range, and the inner logs_for_filter branches on\u00a0which.<\/p>\n \/\/ reth: crates\/rpc\/rpc\/src\/eth\/filter.rs let info = self.provider().chain_info()?; if let Some(t) = to && t > info.best_number { let (from_block_number, to_block_number) = \/\/ Check if the requested range overlaps with pruned history (EIP-4444) self.get_logs_in_block_range(filter, from_block_number, to_block_number, limits) The block-hash branch already knows its block, so it fetches that block\u2019s receipts (cache first, storage on a miss), applies the same pruned-history check, and jumps straight to the exact-matching step every path eventually reaches. No range to resolve, no bloom\u00a0scan.<\/p>\n The pending branch applies when fromBlock is pending, referring to a block that has not yet become part of the canonical chain. Reth uses the pending block it builds locally when one is available.<\/p>\n Ours is the third branch, and it is the only one with work to do before the search can start. Its problem is that \u201ca range\u201d arrived in an unresolved form. Both bounds are optional, and each can be a tag like latest or earliest rather than a number. So it turns the unresolved range into two concrete integers. chain_info() supplies the current head. convert_block_number resolves each bound. A number stays a number, latest becomes the head, and earliest becomes the first block the node has. Then come the boundary rules. These are worth knowing as a user because they are asymmetric:<\/p>\n A toBlock beyond the head is an error (BlockRangeExceedsHead), you asked for blocks that don’t exist\u00a0yet.A fromBlock beyond the head silently returns an empty Vec, the whole search starts past the known\u00a0chain.A range starting before the node\u2019s earliest stored block fails with PrunedHistoryUnavailable, rather than silently searching only the part that survived\u00a0pruning.An omitted bound is not what you\u2019d assume. get_filter_block_range fills a missing fromBlock with the current head, and a missing toBlock with the head too. A filter with no block bounds is a query over the latest block, not over all of\u00a0history.<\/p>\n Our numeric bounds pass every check unchanged:<\/p>\n from_block_number = 25_000_000 The unresolved range is now exactly three concrete\u00a0blocks.<\/p>\n The resolved range enters get_logs_in_block_range, which does two things: checks the guardrails, and moves the work somewhere it’s allowed to be\u00a0slow.<\/p>\n \/\/ reth: crates\/rpc\/rpc\/src\/eth\/filter.rs \/\/ perform boundary checks first if let Some(max_blocks_per_filter) = let (tx, rx) = oneshot::channel(); rx.await.map_err(|_| EthFilterError::InternalError)? The first two checks reject an inverted range and apply Reth\u2019s configured block-range limit before the scan\u00a0begins.<\/p>\n The scan performs synchronous provider operations such as headers_range and receipts_by_block. Since these storage reads can block, Reth runs get_logs_in_block_range_inner in a separate blocking task instead of directly in the RPC handler. The worker sends its result through tx, while rx.await keeps the handler waiting until the result\u00a0arrives.<\/p>\n Everything from here until the worker returns happens inside get_logs_in_block_range_inner. The complete functional path is shown below; tracing-only logging has been\u00a0omitted.<\/p>\n \/\/ reth: crates\/rpc\/rpc\/src\/eth\/filter.rs \/\/ get current chain tip to determine processing mode \/\/ first collect all headers that match the bloom filter for cached mode decision while let Some(header) = headers_iter.next() { let current_number = header.number();<\/p>\n let block_hash = match headers_iter.peek() { matching_headers.push(SealedHeader::new(header, block_hash)); \/\/ initialize the appropriate range mode based on collected headers \/\/ iterate through the range mode to get receipts and blocks \/\/ size check but only if range is multiple blocks, so we always return all Ok(all_logs) The function has three\u00a0stages:<\/p>\n Use header blooms to find candidate blocks (matching_headers).Fetch the receipts for each candidate (range_mode.next()).Check the real logs, build RPC Log values, and enforce the result limit (append_matching_block_logs and the check after\u00a0it).<\/p>\n The first stage starts with a problem of scale. Logs live inside transaction receipts, and since this is our first mention of receipts, it\u2019s worth saying what one is. A receipt is the record Ethereum produces for every executed transaction: whether it succeeded, how much gas it used, and -what matters here- the complete list of logs that transaction emitted. A block carries one receipt per transaction, so a busy block holds hundreds of them. Reading every receipt of every block in the range would work for our three blocks, but would be hopeless for the thousands a serious indexer asks for. The question phase 1 answers: can irrelevant blocks be eliminated without opening their receipts at\u00a0all?<\/p>\n The block header makes that first elimination possible: it carries a 256-byte summary called the logs bloom<\/strong>. When a block is executed, the address and topics of every emitted log get hashed into this bloom filter. A bloom filter can answer one question, fast: \u201ccould this address or topic appear in this block\u2019s logs?\u201d It can produce false positives -\u201dmaybe\u201d when the truth is no- but never false negatives. If the bloom says our USDC address isn\u2019t in this block, it is not in this block, guaranteed, and the node never needs to open the block\u2019s receipts.<\/p>\n So the scan walks headers, not receipts:<\/p>\n \/\/ reth: crates\/rpc\/rpc\/src\/eth\/filter.rs – the bloom scan, in get_logs_in_block_range_inner while let Some(header) = headers_iter.next() { let current_number = header.number(); matching_headers.push(SealedHeader::new(header, block_hash)); The whole decision for each block is one call, matches_bloom:<\/p>\n \/\/ alloy: crates\/rpc-types-eth\/src\/filter.rs The address must be possible in the bloom, and every constrained topic position must be possible too. One negative result skips the block, so its receipts are never opened. A positive result only keeps the block as a candidate; the bloom cannot confirm that a matching log exists. And it cannot confirm where<\/em> a topic sits: the bloom records that a value appears somewhere in the block\u2019s logs, not in which topic slot. So the positional check that phase 3 runs still has real work to\u00a0do.<\/p>\n Two mechanical details are worth a note. First, batching. BlockRangeInclusiveIter slices the range into chunks of max_headers_range, so no single storage read is unbounded. Memory use is still not constant because every bloom-positive header is kept in matching_headers. Our request is one chunk of\u00a0three.<\/p>\n Second, block hashes. Phase 2 fetches receipts by hash, so each surviving header has to leave the scan with its hash attached (SealedHeader::new). Computing one directly means keccak-hashing the whole header, which is the costly path. This is why the iterator is peekable: a header already stores its parent’s hash, so if the next header is consecutive, its parent_hash is exactly the current block’s hash,\u00a0free.<\/p>\n The flip side of the bloom check is worth stating clearly: if the filter names no address and no topics, there is nothing to check; every bloom passes by default, and every block in the range falls through to phase 2, which is exactly why an unfiltered eth_getLogs over a wide range is such an expensive query to send a\u00a0node.<\/p>\n Phase 1\u2019s output is matching_headers: every block whose bloom says the filter may match, each paired with its hash. For our three-block request, this vector can contain anywhere from zero to three headers. The source code tells us how candidates are selected, but only the actual chain data can tell us which of these three blooms\u00a0pass.<\/p>\n A bloom-positive header is only a candidate. The actual logs remain inside the block\u2019s receipts, so Reth must now obtain one receipt list for every candidate in matching_headers.<\/p>\n To see why this step has two strategies, start with where a Reth node keeps its data. Everything the node retains lives in persistent storage on disk. On top of that, Reth keeps recently touched blocks and receipts in memory, most of them near the chain tip. Memory answers immediately and disk is far slower per block. So the same request for a block\u2019s receipts has a cheap answer for recent blocks and an expensive one for old ones, and it is worth choosing the access path explicitly.<\/p>\n That choice is RangeMode. It performs no additional filtering; it only decides how receipt access is managed. Two\u00a0modes:<\/p>\n \/\/ reth: crates\/rpc\/rpc\/src\/eth\/filter.rs Which mode runs is decided when the RangeMode is built. RangeMode::new computes two values from the resolved range: the number of blocks, and the distance from the range’s end to the chain tip. And passes them to should_use_cached_mode:<\/p>\n \/\/ reth: crates\/rpc\/rpc\/src\/eth\/filter.rs \/\/ Calculate adjusted threshold based on bloom matches block_count <= adjusted_threshold && distance_from_tip <= adjusted_threshold The returned boolean selects the mode: true builds Cached, false builds Range (an empty candidate list goes to Range either\u00a0way).<\/p>\n For our query, the decision is easy: three blocks sitting far behind the chain tip, so RangeMode::new picks RangeBlockMode. Iterating it determines how the rest of the phase proceeds. Each range_mode.next() delegates to RangeBlockMode::next, which groups consecutive candidate headers and chooses how to fetch their receipts. With only three candidates, the request stays on the sequential path: check the cache first, then load whatever’s missing through receipts_by_block. Whichever mode runs, next() hands back the same bundle, ReceiptBlockResult:<\/p>\n \/\/ reth: crates\/rpc\/rpc\/src\/eth\/filter.rs Two of the three fields are unconditional: the receipts and the header. The third, recovered_block, is present only when the full block was already in memory; otherwise it is None. That optional block is exactly what Phase 3 reaches for when it needs a matching log’s transaction hash. So whether it arrives as Some or None sets the cost of that\u00a0lookup.<\/p>\n Every path has now converged: for each block that made it this far, get_logs_in_block_range_inner destructures each ReceiptBlockResult that next() returns -pulling out the recovered_block we just saw- and calls the append_matching_block_logs function:<\/p>\n \/\/ reth: crates\/rpc\/rpc\/src\/eth\/filter.rs – the call in get_logs_in_block_range_inner That second argument is the recovered_block from phase 2, mapped into an either-or: ProviderOrBlock::Block when the block came back in memory, ProviderOrBlock::Provider -a handle to storage- when it didn’t. The function needs it for a single field, and the variant it gets sets the price of filling that\u00a0field.<\/p>\n Inside, the function is two nested loops for every receipt in the block, and every log in the\u00a0receipt:<\/p>\n \/\/ reth: crates\/rpc\/rpc-eth-types\/src\/logs_utils.rs<\/p>\n \/\/\/ Appends all matching logs of a block’s receipts. \/\/ Tracks the index of a log in the entire block. \/\/ Lazy loaded number of the first transaction in the block. \/\/ Iterate over receipts and append matching logs. for log in receipt.logs() { \/\/ This is safe because Transactions and Receipts have the same Some(*transaction.tx_hash()) let log = Log { Filter::matches is where the bloom\u2019s false positives die:<\/p>\n \/\/ alloy: crates\/rpc-types-eth\/src\/filter.rs The address is checked against the address set, and each topic against its positional filter -the same positional semantics from the start of the\u00a0article.<\/p>\n Once a log clears matches, the function looks up its transaction hash. This is where the two ProviderOrBlock variants diverge. If phase 2 delivered the block, the answer is an array index: the receipt_idx-th transaction’s hash. If not, storage answers in two\u00a0reads.<\/p>\n Reth\u2019s database numbers every transaction it stores with one global sequence, and a block\u2019s transactions sit consecutively in it. block_body_indices fetches where the block’s transactions begin (first_tx_num), and first_tx_num + receipt_idx gives the transaction’s global number, which transaction_by_id resolves to the transaction itself.<\/p>\n Both lookups are lazy: the hash is fetched only for the first matching log in a receipt (transaction_hash starts as None, and later matches in the same receipt reuse it), while first_tx_num is fetched once per block and reused across receipts.<\/p>\n Once transaction_hash has been resolved, the log that passed matches becomes the RPC Log you will eventually receive. The inner loop adds it to all_logs with all_logs.push(Log {\u00a0\u2026 }). Every other field comes directly from values the loop already\u00a0holds:<\/p>\n inner: the log as the contract emitted it; address, topics,\u00a0data.block_hash, block_number, block_timestamp: straight from values already in\u00a0hand.transaction_index: receipt_idx, because receipts are ordered the same way as the block\u2019s transactions.log_index: the log\u2019s absolute position among all logs in the block. This is why the counter increments outside the if. Non-matching logs still occupy positions.removed: false here; it turns true on the eth_subscribe(“logs”) path, where a reorg replays the logs of the blocks it\u00a0undid.<\/p>\n After append_matching_block_logs returns, the worker applies the second configured limit, max_logs_per_response:<\/p>\n \/\/ reth: crates\/rpc\/rpc\/src\/eth\/filter.rs Three details define its behavior:<\/p>\n It runs after each block, not each log, so a single block\u2019s matching logs are never truncated mid-block.A query for exactly one block is exempt entirely. The is_multi_block_range guard provides this exemption.On overflow, Reth returns a QueryExceedsMaxResults error rather than a partial vector. The error is unusually helpful because it carries from_block and the last block number that still fit. If the very first block already overflowed, it carries that block number instead. A retry can always fetch that block because single-block queries are exempt. In either case, the client is told exactly which smaller range to try\u00a0next.<\/p>\n When the iterator runs dry, the loop ends and get_logs_in_block_range returns Ok(all_logs). For our request, every USDC Transfer log in blocks 25,000,000 through 25,000,002.<\/p>\n When the scan finishes, the result travels back through the same call chain in reverse. Because the scan ran in a separate blocking task, its first hop is through the oneshot channel: the worker sends the result through tx, and the rx.await inside get_logs_in_block_range receives\u00a0it.<\/p>\n From there, each function that awaited the next one resumes and returns the result upward: get_logs_in_block_range returns to the inner logs_for_filter, which returns through the forwarding wrapper and finally reaches the public logs handler. An error follows the same path as a successful Vec<Log>.<\/p>\n Once the handler returns, Reth\u2019s method-specific work is complete. jsonrpsee turns the returned value into a JSON-RPC response and writes it to HTTP, sending the call back across the boundary. Somewhere on the other side, the future we left in AwaitingResponse is about to be polled one last\u00a0time.<\/p>\n Now the third movement reverses the first: raw JSON becomes typed Rust\u00a0again.<\/p>\n The reply arrives on the same HTTP connection, and we are still inside do_reqwest, parked since the POST:\u00a0.send().await resolves once the response headers arrive, and resp.bytes().await reads the\u00a0body:<\/p>\n \/\/ crates\/transport-http\/src\/reqwest_transport.rs let body = resp.bytes().await.map_err(TransportErrorKind::custom)?;<\/p>\n serde_json::from_slice(&body) What does that parse produce? Not your logs, not yet. from_slice returns a ResponsePacket-a thin wrapper that is either Single(Response) or Batch(Vec<Response>):<\/p>\n \/\/ crates\/json-rpc\/src\/packet.rs We sent one request, so ours is Single. Here is the Response:<\/p>\n \/\/ crates\/json-rpc\/src\/response\/mod.rs \/\/ crates\/json-rpc\/src\/response\/payload.rs The payload is classified as Success or Failure while the response is parsed. Its default type is Box<RawValue>, so serde validates the result JSON but leaves it as raw text. This is the first stage of deserialization. The response structure becomes typed, while the result remains\u00a0raw.<\/p>\n The transport cannot convert the result into Vec<Log> because it does not know the expected response type. RpcCall does, and try_deserialize_ok performs the second stage. Batch requests and multiplexed transports use the response id for routing. This single HTTP path discards it without checking.<\/p>\n This parsed ResponsePacket resolves the second wait -the fut.poll(cx) that had been sitting Pending in AwaitingResponse. From here, two transformations turn the raw reply into your Vec<Log>, undoing the two that built the request on the way\u00a0out.<\/p>\n First transformation:<\/strong> convert the response into a Result.<\/strong> Back in the inner poll, the resolved future hands the packet to transform_response:<\/p>\n \/\/ crates\/rpc-client\/src\/call.rs And transform_response is a match on the enum we just\u00a0saw:<\/p>\n \/\/ crates\/json-rpc\/src\/result.rs transform_response converts a failure payload into a Rust Err carrying an RpcError; a successful one stays raw JSON as Box<RawValue>. The\u00a0.. pattern discards the response id. If Reth rejected the query, for example because the block range exceeded the node\u2019s limit, the caller\u2019s\u00a0? propagates that error. The inner state machine then sets itself to Complete. On success, it returns the raw result to the outer poll, where it is deserialized into Vec<Log>.<\/p>\n Second transformation: parse the payload. We now return to the outer poll, the line we set aside in \u201cSending the Request\u201d:<\/p>\n \/\/ crates\/rpc-client\/src\/call.rs try_deserialize_ok is where the raw JSON finally becomes the type you asked for. Here is the function in\u00a0full:<\/p>\n \/\/ crates\/json-rpc\/src\/result.rs \/\/ Fast path: the caller wants the already-owned `Box<RawValue>` back unchanged. Hand it let _guard = debug_span!(“deserialize_response”, ty=%std::any::type_name::<T>()).entered(); result? first propagates anything that has already gone wrong. A transport failure or an error response from the node exits here, and your\u00a0? sees it. What remains is the result payload. It is still raw JSON text that has been syntax-checked but not yet deserialized into Vec<Log>.<\/p>\n Two lines turn that text into your logs. json.borrow().get() exposes the raw value as a plain &str without parsing it. serde_json::from_str then reads it as the response type get_logs declared, the Vec<Log> carried by RpcCall. The transport could not perform this parse because only the caller knows the expected result type. The hex strings become typed addresses, hashes, and numbers inside the same shared Log type that Reth serialized.<\/p>\n If the text refuses the type, the node has sent a shape that the struct cannot accept. In that case, deser_err attaches the failing JSON to the error. A deserialization failure from Alloy shows you the exact text it could not deserialize, which turns \u201csomething didn\u2019t parse\u201d into a bug report that writes\u00a0itself.<\/p>\n (The fast path above serves callers that request Box<RawValue> directly. This is the raw, unparsed JSON. Alloy hands back the value it already owns. Our get_logs never takes this\u00a0path.)<\/p>\n One step remains in the outer\u00a0poll:<\/p>\n Ready(resp.map(this.map.take().expect(“polled after completion”)))<\/p>\n map is a final conversion applied to the response. It provides a chance to reshape the response before it reaches you. get_logs does not need this conversion. The Vec<Log> is already the type you want, so its map is the identity function. The value passes through unchanged.<\/p>\n The outer poll returns Ready, the\u00a0.await inside the default get_logs completes, and get_logs returns the Vec<Log> to its\u00a0caller.<\/p>\n Run against a node, our three-block request returns 187 logs. Here is an example log from the\u00a0result:<\/p>\n Log { At the call site, all of this comes back to the line we started\u00a0with:<\/p>\n let logs = provider.get_logs(&filter).await?;<\/p>\n The path has the same shape in both directions. A typed Filter becomes JSON, crosses into another process, and becomes a typed Filter again. Reth then uses it to check block headers, load receipts, and match logs in stages. The result returns as JSON and becomes a typed Vec<Log>. Alloy builds and sends the call. Reth answers\u00a0it.<\/p>\n A single\u00a0.await covers this whole path. It is not one hidden operation but a series of smaller steps. Once we can see those steps, eth_getLogs no longer looks like magic. A slow query, an empty result, a range-limit error, a transport failure, and a deserialization failure can arise at different points on the same trip. Now we know where to look for each\u00a0one.<\/p>\n The Lifecycle of an Ethereum RPC Call: From Alloy to Reth<\/a> was originally published in Coinmonks<\/a> on Medium, where people are continuing the conversation by highlighting and responding to this story.<\/p>","protected":false},"excerpt":{"rendered":" The Lifecycle of an Ethereum RPC Request: From Alloy to\u00a0Reth let logs = provider.get_logs(&filter).await?; While using Alloy, I kept wondering what a single line like the one above actually does. So I read the relevant parts of Alloy and Reth and wrote down what I\u00a0found. We follow one call, eth_getLogs, over HTTP, from the typed […]<\/p>\n","protected":false},"author":0,"featured_media":206520,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[2],"tags":[],"class_list":["post-206519","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-interesting"],"_links":{"self":[{"href":"https:\/\/mycryptomania.com\/index.php?rest_route=\/wp\/v2\/posts\/206519"}],"collection":[{"href":"https:\/\/mycryptomania.com\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/mycryptomania.com\/index.php?rest_route=\/wp\/v2\/types\/post"}],"replies":[{"embeddable":true,"href":"https:\/\/mycryptomania.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=206519"}],"version-history":[{"count":0,"href":"https:\/\/mycryptomania.com\/index.php?rest_route=\/wp\/v2\/posts\/206519\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/mycryptomania.com\/index.php?rest_route=\/wp\/v2\/media\/206520"}],"wp:attachment":[{"href":"https:\/\/mycryptomania.com\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=206519"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/mycryptomania.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=206519"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/mycryptomania.com\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=206519"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}Overview<\/h3>\n
.address(address!(“A0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48”)) \/\/ USDC
.from_block(0x17d7840u64)
.to_block(0x17d7842u64)
.event_signature(keccak256(“Transfer(address,address,uint256)”));<\/p>\n
“method”: “eth_getLogs”,
“params”: [
{
“fromBlock”: “0x17d7840”,
“toBlock”: “0x17d7842”,
“address”: “0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48”,
“topics”: [
“0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef”
]
}
],
“id”: 0,
“jsonrpc”: “2.0”
}<\/p>\n
“jsonrpc”: “2.0”,
“id”: 0,
“result”: [
{
“address”: “0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48”,
“topics”: [“0xddf252ad…”, “0x…”, “0x…”],
“data”: “0x…”,
“blockHash”: “0x…”,
“blockNumber”: “0x17d7840”,
“blockTimestamp”: “0x…”,
“transactionHash”: “0x…”,
“transactionIndex”: “0x2”,
“logIndex”: “0x5”,
“removed”: false
}
]
}<\/p>\nWhat the Filter\u00a0Says<\/h3>\n
pub struct Filter {
pub block_option: FilterBlockOption,
pub address: FilterSet<Address>,
pub topics: [Topic; 4],
}<\/p>\n
topics[1] = from
topics[2] = to
data = value<\/p>\n
“topics”: [
“0xddf252ad…”,
null,
“0x000000000000000000000000…”
]
}<\/p>\nHow Alloy Serializes and Sends the\u00a0Call<\/h3>\n
Provider<\/h3>\n
pub trait Provider<N: Network = Ethereum>: Send + Sync {
\/\/\/ Returns the root provider.
fn root(&self) -> &RootProvider<N>;<\/p>\n
#[inline]
fn client(&self) -> ClientRef<‘_> {
self.root().client()
}<\/p>\n
async fn get_logs(&self, filter: &Filter) -> TransportResult<Vec<Log>> {
self.client().request(“eth_getLogs”, (filter,)).await
}
}<\/p>\nRootProvider<\/h3>\n
pub struct RootProvider<N: Network = Ethereum> {
\/\/\/ The inner state of the root provider.
pub(crate) inner: Arc<RootProviderInner<N>>,
}<\/p>\n
client: RpcClient,
heart: OnceLock<HeartbeatHandle>,
_network: PhantomData<N>,
}<\/p>\nRpcClient<\/h3>\n
pub struct RpcClientInner {
\/\/\/ The underlying transport.
pub(crate) transport: BoxTransport,
\/\/\/ The next request ID to use.
pub(crate) id: AtomicU64,
\/\/ …
}<\/p>\nTransport<\/h3>\n
pub struct BoxTransport {
inner: Box<dyn CloneTransport>,
}<\/p>\nAlloy\u2019s Serialization<\/h3>\n
async fn get_logs(&self, filter: &Filter) -> TransportResult<Vec<Log>> {
self.client().request(“eth_getLogs”, (filter,)).await
}<\/p>\n
\/\/\/ Prepares an [`RpcCall`].
\/\/\/
\/\/\/ This function reserves an ID for the request, however the request is not sent.
\/\/\/ To send a request, await the returned [`RpcCall`].
\/\/\/
\/\/\/ # Note
\/\/\/
\/\/\/ Serialization is done lazily. It will not be performed until the call is awaited.
pub fn request<Params: RpcSend, Resp: RpcRecv>(
&self,
method: impl Into<Cow<‘static, str>>,
params: Params,
) -> RpcCall<Params, Resp> {
let request = self.make_request(method, params);
RpcCall::new(request, self.transport.clone())
}<\/p>\n
&self,
method: impl Into<Cow<‘static, str>>,
params: Params,
) -> Request<Params> {
Request::new(method, self.next_id(), params)
}<\/p>\nSending the\u00a0Request<\/h3>\n
fn poll(self: Pin<&mut Self>, cx: &mut task::Context<‘_>) -> task::Poll<Self::Output> {
let this = self.get_mut();
let resp = try_deserialize_ok(ready!(this.state.poll_unpin(cx)));
Ready(resp.map(this.map.take().expect(“polled after completion”)))
}<\/p>\n
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<‘_>) -> task::Poll<Self::Output> {
loop {
match self.as_mut().project() {
CallStateProj::Prepared { connection, request } => {
if let Err(e) =
task::ready!(Service::<RequestPacket>::poll_ready(connection, cx))
{
self.set(Self::Complete);
return Ready(RpcResult::Err(e));
}<\/p>\n
if tracing::enabled!(tracing::Level::TRACE) {
trace!(?request, “sending request”);
} else {
debug!(method=%request.meta.method, id=%request.meta.id, “sending request”);
}
let request = request.serialize();
let fut = match request {
Ok(request) => {
trace!(request=%request.serialized(), “serialized request”);
connection.call(request.into())
}
Err(err) => {
trace!(?err, “failed to serialize request”);
self.set(Self::Complete);
return Ready(RpcResult::Err(TransportError::ser_err(err)));
}
};
self.set(Self::AwaitingResponse { fut });
}
CallStateProj::AwaitingResponse { fut } => {
let res = match task::ready!(fut.poll(cx)) {
Ok(ResponsePacket::Single(res)) => Ready(transform_response(res)),
Err(e) => Ready(RpcResult::Err(e)),
Ok(ResponsePacket::Batch(_)) => {
Ready(RpcResult::Err(TransportErrorKind::custom_str(
“received batch response from single request”,
)))
}
};
self.set(Self::Complete);
return res;
}
CallStateProj::Complete => {
panic!(“Polled after completion”);
}
}
}
}<\/p>\nThe Prepared\u00a0arm<\/h4>\n
if let Err(e) =
task::ready!(Service::<RequestPacket>::poll_ready(connection, cx))
{
self.set(Self::Complete);
return Ready(RpcResult::Err(e));
}<\/p>\n
fn poll_ready(&mut self, _cx: &mut task::Context<‘_>) -> task::Poll<Result<(), Self::Error>> {
\/\/ `reqwest` always returns `Ok(())`.
task::Poll::Ready(Ok(()))
}<\/p>\n
let request = request.take().expect(“no request”);
let request = request.serialize();
let fut = match request {
Ok(request) => connection.call(request.into()),
Err(err) => {
self.set(Self::Complete);
return Ready(RpcResult::Err(TransportError::ser_err(err)));
}
};
self.set(Self::AwaitingResponse { fut });<\/p>\n
pub fn serialize(self) -> serde_json::Result<SerializedRequest> {
let request = serde_json::value::to_raw_value(&self)?;
Ok(SerializedRequest { meta: self.meta, request })
}<\/p>\n
\/\/ manually implemented to avoid adding a type for the protocol-required
\/\/ `jsonrpc` field
impl<Params> Serialize for Request<Params>
where
Params: RpcSend,
{
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
let sized_params = std::mem::size_of::<Params>() != 0;<\/p>\n
map.serialize_entry(“method”, &self.meta.method[..])?;<\/p>\n
if sized_params {
map.serialize_entry(“params”, &self.params)?;
}<\/p>\n
map.serialize_entry(“jsonrpc”, “2.0”)?;
map.end()
}
}<\/p>\n
impl Service<RequestPacket> for Http<reqwest::Client> {
type Response = ResponsePacket;
type Error = TransportError;
type Future = TransportFut<‘static>;<\/p>\n
fn call(&mut self, req: RequestPacket) -> Self::Future {
let this = self.clone();
let span = debug_span!(“ReqwestTransport”, url = %this.url);
Box::pin(this.do_reqwest(req).instrument(span.or_current()))
}
}<\/p>\nThe AwaitingResponse arm<\/h4>\n
let res = match task::ready!(fut.poll(cx)) {
Ok(ResponsePacket::Single(res)) => Ready(transform_response(res)),
Err(e) => Ready(RpcResult::Err(e)),
Ok(ResponsePacket::Batch(_)) => {
Ready(RpcResult::Err(TransportErrorKind::custom_str(
“received batch response from single request”,
)))
}
};
self.set(Self::Complete);<\/p>\n
async fn do_reqwest(self, req: RequestPacket) -> TransportResult<ResponsePacket> {
let resp = self
.client
.post(self.url)
.json(&req)
.headers(req.headers())
.send()
.await
.map_err(TransportErrorKind::custom)?;
let status = resp.status();<\/p>\n
\/\/ the status code, as we want to return the error in the body
\/\/ if there is one.
let body = resp.bytes().await.map_err(TransportErrorKind::custom)?;<\/p>\n
if let Some(response) = crate::json_rpc_error_response(&body) {
return Ok(response);
}<\/p>\n
status.as_u16(),
String::from_utf8_lossy(&body).into_owned(),
));
}<\/p>\n
\/\/ the body as a string in the error. The conversion to String
\/\/ is lossy and may not cover all the bytes in the body.
serde_json::from_slice(&body)
.map_err(|err| TransportError::deser_err(err, String::from_utf8_lossy(&body)))
}<\/p>\nThe Complete\u00a0arm<\/h4>\n
panic!(“Polled after completion”);
}<\/p>\nInside Reth<\/h3>\n
Where the Request\u00a0Arrives<\/h3>\n
#[cfg_attr(not(feature = “client”), rpc(server, namespace = “eth”))]
#[cfg_attr(feature = “client”, rpc(server, client, namespace = “eth”))]
pub trait EthFilterApi<T: RpcObject> {
\/\/ …<\/p>\n
#[method(name = “getLogs”)]
async fn logs(&self, filter: Filter) -> RpcResult<Vec<Log>>;
}<\/p>\n
#[async_trait]
impl<Eth> EthFilterApiServer<RpcTransaction<Eth::NetworkTypes>> for EthFilter<Eth>
where
Eth: FullEthApiTypes + RpcNodeCoreExt + LoadReceipt + EthBlocks + ‘static,
{
\/\/ …<\/p>\n
async fn logs(&self, filter: Filter) -> RpcResult<Vec<Log>> {
trace!(target: “rpc::eth”, “Serving eth_getLogs”);
Ok(self.logs_for_filter(filter, self.inner.query_limits).await?)
}
}<\/p>\n
&self,
filter: Filter,
limits: QueryLimits,
) -> Result<Vec<Log>, EthFilterError> {
self.inner.clone().logs_for_filter(filter, limits).await
}<\/p>\nResolving the Block\u00a0Range<\/h3>\n
async fn logs_for_filter(
self: Arc<Self>,
filter: Filter,
limits: QueryLimits,
) -> Result<Vec<Log>, EthFilterError> {
match filter.block_option {
FilterBlockOption::AtBlockHash(block_hash) => { \/* single-block path omitted *\/ }
FilterBlockOption::Range { from_block, to_block } => {
if from_block.is_some_and(|b| b.is_pending()) {
\/* pending-block path omitted *\/
}<\/p>\n
let start_block = info.best_number;
let from = from_block
.map(|num| self.provider().convert_block_number(num))
.transpose()?
.flatten();
let to = to_block
.map(|num| self.provider().convert_block_number(num))
.transpose()?
.flatten();<\/p>\n
return Err(EthFilterError::BlockRangeExceedsHead {
requested: t, head: info.best_number,
});
}
if let Some(f) = from && f > info.best_number {
return Ok(Vec::new());
}<\/p>\n
logs_utils::get_filter_block_range(from, to, start_block, info)?;<\/p>\n
let earliest_block = self.provider().earliest_block_number()?;
if from_block_number < earliest_block {
return Err(EthApiError::PrunedHistoryUnavailable.into());
}<\/p>\n
.await
}
}
}<\/p>\n
to_block_number = 25_000_002<\/p>\nRange Limits and a Blocking\u00a0Task<\/h3>\n
async fn get_logs_in_block_range(
self: Arc<Self>,
filter: Filter,
from_block: u64,
to_block: u64,
limits: QueryLimits,
) -> Result<Vec<Log>, EthFilterError> {
trace!(target: “rpc::eth::filter”, from=from_block, to=to_block, ?filter, “finding logs in range”);<\/p>\n
if to_block < from_block {
return Err(EthFilterError::InvalidBlockRangeParams);
}<\/p>\n
limits.max_blocks_per_filter.filter(|limit| to_block – from_block > *limit)
{
return Err(EthFilterError::QueryExceedsMaxBlocks(max_blocks_per_filter));
}<\/p>\n
let this = self.clone();
self.task_spawner.spawn_blocking_task(async move {
let res =
this.get_logs_in_block_range_inner(&filter, from_block, to_block, limits).await;
let _ = tx.send(res);
});<\/p>\n
}<\/p>\nThe Scan in Three\u00a0Phases<\/h3>\n
async fn get_logs_in_block_range_inner(
self: Arc<Self>,
filter: &Filter,
from_block: u64,
to_block: u64,
limits: QueryLimits,
) -> Result<Vec<Log>, EthFilterError> {
let mut all_logs = Vec::new();
let mut matching_headers = Vec::new();<\/p>\n
let chain_tip = self.provider().best_block_number()?;<\/p>\n
for (from, to) in
BlockRangeInclusiveIter::new(from_block..=to_block, self.max_headers_range)
{
let headers = self.provider().headers_range(from..=to)?;
let mut headers_iter = headers.into_iter().peekable();<\/p>\n
if !filter.matches_bloom(header.logs_bloom()) {
continue;
}<\/p>\n
Some(next_header) if next_header.number() == current_number + 1 => {
\/\/ Headers are consecutive, use the more efficient parent_hash
next_header.parent_hash()
}
_ => {
\/\/ Headers not consecutive or last header, calculate hash
header.hash_slow()
}
};<\/p>\n
}
}<\/p>\n
let mut range_mode = RangeMode::new(
self.clone(),
matching_headers,
from_block,
to_block,
self.max_headers_range,
chain_tip,
);<\/p>\n
while let Some(ReceiptBlockResult { receipts, recovered_block, header }) =
range_mode.next().await?
{
let num_hash = header.num_hash();
append_matching_block_logs(
&mut all_logs,
recovered_block
.map(ProviderOrBlock::Block)
.unwrap_or_else(|| ProviderOrBlock::Provider(self.provider())),
filter,
num_hash,
&receipts,
false,
header.timestamp(),
)?;<\/p>\n
\/\/ logs of a single block
let is_multi_block_range = from_block != to_block;
if let Some(max_logs_per_response) = limits.max_logs_per_response
&& is_multi_block_range
&& all_logs.len() > max_logs_per_response
{
let retry_to_block =
if num_hash.number == from_block { from_block } else { num_hash.number – 1 };
return Err(EthFilterError::QueryExceedsMaxResults {
max_logs: max_logs_per_response,
from_block,
to_block: retry_to_block,
});
}
}<\/p>\n
}<\/p>\nPhase 1: The bloom\u00a0scan<\/h3>\n
for (from, to) in BlockRangeInclusiveIter::new(from_block..=to_block, self.max_headers_range) {
let headers = self.provider().headers_range(from..=to)?;
let mut headers_iter = headers.into_iter().peekable();<\/p>\n
if !filter.matches_bloom(header.logs_bloom()) {
continue;
}<\/p>\n
let block_hash = match headers_iter.peek() {
Some(next_header) if next_header.number() == current_number + 1 => {
\/\/ Headers are consecutive, use the more efficient parent_hash
next_header.parent_hash()
}
_ => {
\/\/ Headers not consecutive or last header, calculate hash
header.hash_slow()
}
};<\/p>\n
}
}<\/p>\n
pub fn matches_bloom(&self, bloom: Bloom) -> bool {
self.address_bloom_filter().matches(bloom)
&& self.topics_bloom_filter().iter().all(|topic_bloom| topic_bloom.matches(bloom))
}<\/p>\nPhase 2: Fetching the\u00a0receipts<\/h3>\n
enum RangeMode<
Eth: RpcNodeCoreExt<Provider: BlockIdReader, Pool: TransactionPool>
+ EthApiTypes
+ LoadReceipt
+ EthBlocks
+ ‘static,
> {
\/\/\/ Use cache-based processing for recent blocks
Cached(CachedMode<Eth>),
\/\/\/ Use range-based processing for older blocks
Range(RangeBlockMode<Eth>),
}CachedMode, for small ranges near the chain tip. Each candidate’s receipts are requested through the cache service (get_receipts_and_maybe_block): already in memory, returned immediately; on a miss, the cache service can still fall back to storage. If the full block is also cached, it comes back alongside the receipts.RangeBlockMode, for older or larger ranges, goes to storage. It checks the cache without forcing a fetch, then reads whatever’s missing straight from disk via self.provider().receipts_by_block(…). A very large candidate set is split across parallel blocking tasks that go straight to the provider, their results reassembled in block order so logs never come back out of sequence.<\/p>\n
\/\/\/ Determines whether to use cached mode based on bloom filter matches and range size
const fn should_use_cached_mode(
headers: &[SealedHeader<<Eth::Provider as HeaderProvider>::Header>],
block_count: u64,
distance_from_tip: u64,
) -> bool {
\/\/ Headers are already filtered by bloom, so count equals length
let bloom_matches = headers.len();<\/p>\n
let adjusted_threshold = Self::calculate_adjusted_threshold(block_count, bloom_matches);<\/p>\n
}<\/p>\n
\/\/\/ Helper type for the common pattern of returning receipts, block and the original header that is
\/\/\/ a match for the filter.
struct ReceiptBlockResult<P>
where
P: ReceiptProvider + BlockReader,
{
\/\/\/ We always need the entire receipts for the matching block.
receipts: Arc<Vec<ProviderReceipt<P>>>,
\/\/\/ Block can be optional and we can fetch it lazily when needed.
recovered_block: Option<Arc<reth_primitives_traits::RecoveredBlock<ProviderBlock<P>>>>,
\/\/\/ The header of the block.
header: SealedHeader<<P as HeaderProvider>::Header>,
}<\/p>\nPhase 3: Exact matching, and building your\u00a0Log<\/h3>\n
append_matching_block_logs(
&mut all_logs,
recovered_block
.map(ProviderOrBlock::Block)
.unwrap_or_else(|| ProviderOrBlock::Provider(self.provider())),
filter,
num_hash,
&receipts,
false,
header.timestamp(),
)?;<\/p>\n
\/\/\/ If the log matches, look up the corresponding transaction hash.
pub fn append_matching_block_logs<P>(
all_logs: &mut Vec<Log>,
provider_or_block: ProviderOrBlock<‘_, P>,
filter: &Filter,
block_num_hash: BlockNumHash,
receipts: &[P::Receipt],
removed: bool,
block_timestamp: u64,
) -> Result<(), ProviderError>
where
P: BlockReader<Transaction: SignedTransaction>,
{
if !filter.matches_block(&block_num_hash) {
return Ok(());
}<\/p>\n
let mut log_index: u64 = 0;<\/p>\n
\/\/ This is useful for blocks with multiple matching logs because it
\/\/ prevents re-querying the block body indices.
let mut loaded_first_tx_num = None;<\/p>\n
for (receipt_idx, receipt) in receipts.iter().enumerate() {
\/\/ The transaction hash of the current receipt.
let mut transaction_hash = None;<\/p>\n
if filter.matches(log) {
\/\/ if this is the first match in the receipt’s logs, look up the transaction hash
if transaction_hash.is_none() {
transaction_hash = match &provider_or_block {
ProviderOrBlock::Block(block) => {
block.body().transactions().get(receipt_idx).map(|t| *t.tx_hash())
}
ProviderOrBlock::Provider(provider) => {
let first_tx_num = match loaded_first_tx_num {
Some(num) => num,
None => {
let block_body_indices = provider
.block_body_indices(block_num_hash.number)?
.ok_or(ProviderError::BlockBodyIndicesNotFound(
block_num_hash.number,
))?;
loaded_first_tx_num = Some(block_body_indices.first_tx_num);
block_body_indices.first_tx_num
}
};<\/p>\n
\/\/ keys.
let transaction_id = first_tx_num + receipt_idx as u64;
let transaction =
provider.transaction_by_id(transaction_id)?.ok_or_else(|| {
ProviderError::TransactionNotFound(transaction_id.into())
})?;<\/p>\n
}
};
}<\/p>\n
inner: log.clone(),
block_hash: Some(block_num_hash.hash),
block_number: Some(block_num_hash.number),
transaction_hash,
\/\/ The transaction and receipt index is always the same.
transaction_index: Some(receipt_idx as u64),
log_index: Some(log_index),
removed,
block_timestamp: Some(block_timestamp),
};
all_logs.push(log);
}
log_index += 1;
}
}
Ok(())
}<\/p>\n
pub fn matches(&self, log: &alloy_primitives::Log) -> bool {
if !self.matches_address(log.address) {
return false;
}
self.matches_topics(log.topics())
}<\/p>\n
\/\/ size check but only if range is multiple blocks, so we always return all
\/\/ logs of a single block
let is_multi_block_range = from_block != to_block;
if let Some(max_logs_per_response) = limits.max_logs_per_response
&& is_multi_block_range
&& all_logs.len() > max_logs_per_response
{
let retry_to_block =
if num_hash.number == from_block { from_block } else { num_hash.number – 1 };
return Err(EthFilterError::QueryExceedsMaxResults {
max_logs: max_logs_per_response,
from_block,
to_block: retry_to_block,
});
}<\/p>\nReturning Through the Call\u00a0Chain<\/h3>\n
How Alloy Deserializes the\u00a0Response<\/h3>\n
async fn do_reqwest(
self,
req: RequestPacket,
) -> TransportResult<ResponsePacket> {
\/\/ request sending and status handling omitted<\/p>\n
.map_err(|err| TransportError::deser_err(
err,
String::from_utf8_lossy(&body),
))
}<\/p>\n
pub enum ResponsePacket<Payload = Box<RawValue>, ErrData = Box<RawValue>> {
\/\/\/ A single response.
Single(Response<Payload, ErrData>),
\/\/\/ A batch of responses.
Batch(Vec<Response<Payload, ErrData>>),
}<\/p>\n
pub struct Response<Payload = Box<RawValue>, ErrData = Box<RawValue>> {
pub id: Id,
pub payload: ResponsePayload<Payload, ErrData>,
}<\/p>\n
pub enum ResponsePayload<Payload = Box<RawValue>, ErrData = Box<RawValue>> {
Success(Payload),
Failure(ErrorPayload<ErrData>),
}<\/p>\n
CallStateProj::AwaitingResponse { fut } => {
let res = match task::ready!(fut.poll(cx)) {
Ok(ResponsePacket::Single(res)) => Ready(transform_response(res)),
Err(e) => Ready(RpcResult::Err(e)),
Ok(ResponsePacket::Batch(_)) => {
Ready(RpcResult::Err(TransportErrorKind::custom_str(
“received batch response from single request”,
)))
}
};
self.set(Self::Complete);
return res;
}<\/p>\n
pub fn transform_response<T, E, ErrResp>(response: Response<T, ErrResp>) -> RpcResult<T, E, ErrResp>
where
ErrResp: RpcRecv,
{
match response {
Response { payload: ResponsePayload::Failure(err_resp), .. } => {
Err(RpcError::err_resp(err_resp))
}
Response { payload: ResponsePayload::Success(result), .. } => Ok(result),
}
}<\/p>\n
let resp = try_deserialize_ok(ready!(this.state.poll_unpin(cx)));<\/p>\n
pub fn try_deserialize_ok<J, T, E, ErrResp>(
result: RpcResult<J, E, ErrResp>,
) -> RpcResult<T, E, ErrResp>
where
J: Borrow<RawValue> + ‘static,
T: RpcRecv,
ErrResp: RpcRecv,
{
let json = result?;<\/p>\n
\/\/ over directly, skipping the byte copy and full JSON validation scan that
\/\/ `from_str::<Box<RawValue>>` would repeat over an already-validated value.
if TypeId::of::<J>() == TypeId::of::<Box<RawValue>>()
&& TypeId::of::<T>() == TypeId::of::<Box<RawValue>>()
{
\/\/ SAFETY: `J` and `T` are both `Box<RawValue>`, so this is a no-op reinterpretation.
\/\/ `transmute_copy` stands in for the unstable `transmute_unchecked` (sizes are equal).
let json = std::mem::ManuallyDrop::new(json);
return Ok(unsafe { std::mem::transmute_copy::<J, T>(&json) });
}<\/p>\n
let json = json.borrow().get();
trace!(%json, “deserializing”);
serde_json::from_str(json)
.inspect(|response| trace!(?response, “deserialized”))
.inspect_err(|err| trace!(?err, “failed to deserialize”))
.map_err(|err| RpcError::deser_err(err, json))
}<\/p>\n
inner: Log {
address: 0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48,
data: LogData {
topics: [
0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef,
0x000000000000000000000000000000000004444c5dc75cb358380d2e3de08a90,
0x0000000000000000000000007f54f05635d15cde17a49502fedb9d1803a3be8a,
],
data: 0x00000000000000000000000000000000000000000000000000000000017d0fc6,
},
},
block_hash: Some(
0xf398976165ca4756c77fc6b61111fa1102d431eb03082417ecce38b36308d728,
),
block_number: Some(
25000000,
),
block_timestamp: Some(
1777637363,
),
transaction_hash: Some(
0xd019ec20783bff41af6c78575e5f1bc4018775161d1e82f7913c2c30a8369d35,
),
transaction_index: Some(
8,
),
log_index: Some(
33,
),
removed: false,
}<\/p>\nClosing the\u00a0Loop<\/h3>\n