[Bug] Logs Disappear After a Chain Reorged Until the Contract Deployment

### Component

node-bindings, provider, pubsub

### What version of Alloy are you on?

1.0.42

### Operating System

Linux

### Describe the bug

TLDR:
- Spawn Anvil (block-per-tx or with block time, bug occurs in both cases)
- Deploy contract
- Execute X number of transactions from this contract that emit events
- Reorg the chain up to, but not including the contract deployment block (i.e. reorg the blocks where contract's txs were executed), re-executing the contract transactions
- Transactions get included, but **no logs are found on-chain**.

---

The logs emitted as part of transactions passed in `ReorgOptions::tx_block_pairs` cannot be fetched after the chain reorg is executed.
Either they never get emitted, and this is really an Anvil issue, or `alloy`'s provider is misconfigured and is unable to fetch them post-reorg.

The peculiar thing is that the same bug does not appear when either:
- the block including the first contract transaction is _not reorged_ (only the subsequent ones)
- if no contract transactions were ever executed.

Just in case the repo becomes unavailable in the future, here are the exact repro steps:
1. Create a new cargo project, run: `cargo new --lib reorg-bug`
2. Open the project
3. Open `Cargo.toml`
4. Paste the following:
```toml
[package]
name = "reorg-bug"
version = "0.1.0"
edition = "2024"

[dependencies]
alloy = { version = "1.0.42", features = ["full"] }
alloy-node-bindings = "1.0.42"
tokio = { version = "1.0", features = ["full"] }
anyhow = "1.0"
```
5. Open `src/lib.rs`
6. Paste the following:
```rust
#[cfg(test)]
mod tests {
use alloy::{
providers::{Provider, ProviderBuilder, ext::AnvilApi},
rpc::types::{
Filter,
anvil::{ReorgOptions, TransactionData},
},
sol,
};

use alloy_node_bindings::Anvil;

sol! {
// Built directly with solc 0.8.30+commit.73712a01.Darwin.appleclang
#[sol(rpc, bytecode="608080604052346015576101b0908161001a8239f35b5f80fdfe6080806040526004361015610012575f80fd5b5f3560e01c90816306661abd1461016157508063a87d942c14610145578063d732d955146100ad5763e8927fbc14610048575f80fd5b346100a9575f3660031901126100a9575f5460018101809111610095576020817f7ca2ca9527391044455246730762df008a6b47bbdb5d37a890ef78394535c040925f55604051908152a1005b634e487b7160e01b5f52601160045260245ffd5b5f80fd5b346100a9575f3660031901126100a9575f548015610100575f198101908111610095576020817f53a71f16f53e57416424d0d18ccbd98504d42a6f98fe47b09772d8f357c620ce925f55604051908152a1005b60405162461bcd60e51b815260206004820152601860248201527f436f756e742063616e6e6f74206265206e6567617469766500000000000000006044820152606490fd5b346100a9575f3660031901126100a95760205f54604051908152f35b346100a9575f3660031901126100a9576020905f548152f3fea2646970667358221220471585b420a1ad0093820ff10129ec863f6df4bec186546249391fbc3cdbaa7c64736f6c634300081e0033")]
contract TestCounter {
uint256 public count;

#[derive(Debug)]
event CountIncreased(uint256 newCount);
#[derive(Debug)]
event CountDecreased(uint256 newCount);

function increase() public {
count += 1;
emit CountIncreased(count);
}

function decrease() public {
require(count > 0, "Count cannot be negative");
count -= 1;
emit CountDecreased(count);
}

function getCount() public view returns (uint256) {
return count;
}
}
}

mod block_per_tx {
use super::*;

#[tokio::test]
async fn test_reorg_all_log_emissions() -> anyhow::Result<()> {
// configure Anvil to mint a block per each transaction
let anvil = Anvil::new().try_spawn()?;
let provider = ProviderBuilder::new()
.wallet(anvil.wallet().unwrap())
.connect(anvil.endpoint().as_str())
.await?;

let contract = TestCounter::deploy(provider.clone()).await?;

// initial event emissions
let event_count = 5;
for _ in 0..event_count {
let _ = contract.increase().send().await?.get_receipt().await?;
}

let contract_deployment_block = provider
.get_block(1.into())
.await?
.expect("block should exist");
assert_eq!(1, contract_deployment_block.transactions.len());

// assert that each block has exactly 1 transaction
let last_event_block = contract_deployment_block.number() + event_count;
for i in 2..=last_event_block {
let block = provider
.get_block(i.into())
.await?
.expect("block should exist");
assert_eq!(1, block.transactions.len());
}

// assert the total number of logs in the whole chain
let log_filter = &Filter::new().from_block(0);

let logs = provider.get_logs(log_filter).await?;
assert_eq!(event_count, logs.len() as u64);

// reorg the last 5 blocks, re-emitting only 3 events
let reorg_depth = event_count;
let new_event_count = 3;

let tx_block_pairs = (0..new_event_count)
.map(|_| {
let tx = contract.increase().into_transaction_request();
(TransactionData::JSON(tx), 0)
})
.collect();

let opts = ReorgOptions {
depth: reorg_depth,
tx_block_pairs,
};
provider.anvil_reorg(opts).await?;

// after the reorg, the block when contract was deployed should be unchanged
let post_reorg_contract_deployment_block = provider
.get_block(contract_deployment_block.number().into())
.await?
.expect("block should exist");
assert_eq!(
contract_deployment_block.hash(),
post_reorg_contract_deployment_block.hash()
);
assert_eq!(1, post_reorg_contract_deployment_block.transactions.len());

// the next block should contain all of the new transactions
let txs_block = provider
.get_block(2.into())
.await?
.expect("block should exist");
assert_eq!(new_event_count, txs_block.transactions.len());

// other blocks should contain no transactions
for i in 3..last_event_block {
let block = provider
.get_block(i.into())
.await?
.expect("block should exist");
assert!(block.transactions.is_empty());
}

// reassert the number of logs in the whole chain
let logs = provider.get_logs(log_filter).await?;
assert_eq!(new_event_count, logs.len()); // FAIL: logs.len() somehow equals 0 (zero)

Ok(())
}

#[tokio::test]
async fn test_reorg_all_log_emissions_but_first() -> anyhow::Result<()> {
// configure Anvil to mint a block per each transaction
let anvil = Anvil::new().try_spawn()?;
let provider = ProviderBuilder::new()
.wallet(anvil.wallet().unwrap())
.connect(anvil.endpoint().as_str())
.await?;

let contract = TestCounter::deploy(provider.clone()).await?;

// initial event emissions
let event_count = 5;
for _ in 0..event_count {
let _ = contract.increase().send().await?.get_receipt().await?;
}

let contract_deployment_block = provider
.get_block(1.into())
.await?
.expect("block should exist");
assert_eq!(1, contract_deployment_block.transactions.len());

let last_event_block = contract_deployment_block.number() + event_count;

// assert that each block has exactly 1 transaction
for i in 2..=last_event_block {
let block = provider
.get_block(i.into())
.await?
.expect("block should exist");
assert_eq!(1, block.transactions.len());
}

// assert the total number of logs in the whole chain
let log_filter = &Filter::new().from_block(0);

let logs = provider.get_logs(log_filter).await?;
assert_eq!(event_count, logs.len() as u64);

// reorg the last 4 blocks, re-emitting only 3 events
let reorg_depth = event_count - 1;
let new_event_count = 3;

let tx_block_pairs = (0..new_event_count)
.map(|_| {
let tx = contract.increase().into_transaction_request();
(TransactionData::JSON(tx), 0)
})
.collect();

let opts = ReorgOptions {
depth: reorg_depth,
tx_block_pairs,
};
provider.anvil_reorg(opts).await?;

// after the reorg, the block when contract was deployed should be unchanged
let post_reorg_contract_deployment_block = provider
.get_block(contract_deployment_block.number().into())
.await?
.expect("block should exist");
assert_eq!(
contract_deployment_block.hash(),
post_reorg_contract_deployment_block.hash()
);
assert_eq!(1, post_reorg_contract_deployment_block.transactions.len());

// after the reorg, the block with the first log emission should be unchanged
let txs_block = provider
.get_block(2.into())
.await?
.expect("block should exist");
assert_eq!(1, txs_block.transactions.len());

// the next block should contain all of the new transactions
let txs_block = provider
.get_block(3.into())
.await?
.expect("block should exist");
assert_eq!(new_event_count, txs_block.transactions.len());

// other blocks should contain no transactions
for i in 4..last_event_block {
let block = provider
.get_block(i.into())
.await?
.expect("block should exist");
assert!(block.transactions.is_empty());
}

// reassert the number of logs in the whole chain is:
// all the newly emitted logs + the 1 old log
let logs = provider.get_logs(log_filter).await?;
assert_eq!(new_event_count + 1, logs.len());

Ok(())
}

#[tokio::test]
async fn test_reorg_no_initial_logs() -> anyhow::Result<()> {
// configure Anvil to mint a block per each transaction
let anvil = Anvil::new().try_spawn()?;
let provider = ProviderBuilder::new()
.wallet(anvil.wallet().unwrap())
.connect(anvil.endpoint().as_str())
.await?;

let contract = TestCounter::deploy(provider.clone()).await?;

let contract_deployment_block = provider
.get_block(1.into())
.await?
.expect("block should exist");
assert_eq!(1, contract_deployment_block.transactions.len());

// mint empty blocks
let block_count = 5;
provider.anvil_mine(Some(block_count), None).await?;

// reorg the last 5 blocks, re-emitting only 3 events
let reorg_depth = block_count;
let new_event_count = 3;

let tx_block_pairs = (0..new_event_count)
.map(|_| {
let tx = contract.increase().into_transaction_request();
(TransactionData::JSON(tx), 0)
})
.collect();

let opts = ReorgOptions {
depth: reorg_depth,
tx_block_pairs,
};
provider.anvil_reorg(opts).await?;

// after the reorg, the block when contract was deployed should be unchanged
let post_reorg_contract_deployment_block = provider
.get_block(contract_deployment_block.number().into())
.await?
.expect("block should exist");
assert_eq!(
contract_deployment_block.hash(),
post_reorg_contract_deployment_block.hash()
);
assert_eq!(1, post_reorg_contract_deployment_block.transactions.len());

// the next block should contain all of the new transactions
let txs_block = provider
.get_block(2.into())
.await?
.expect("block should exist");
assert_eq!(new_event_count, txs_block.transactions.len());

// other blocks should contain no transactions
for i in 3..7 {
let block = provider
.get_block(i.into())
.await?
.expect("block should exist");
assert!(block.transactions.is_empty());
}

// assert the total number of logs in the whole chain
let logs = provider.get_logs(&Filter::new().from_block(0)).await?;
assert_eq!(new_event_count, logs.len()); // FAIL: logs.len() somehow equals 0 (zero)

Ok(())
}
}

mod with_block_time {
use super::*;

#[tokio::test]
async fn test_reorg_all_log_emissions() -> anyhow::Result<()> {
// configure Anvil to mint a block per each transaction
let anvil = Anvil::new().block_time_f64(0.1).try_spawn()?;
let provider = ProviderBuilder::new()
.wallet(anvil.wallet().unwrap())
.connect(anvil.endpoint().as_str())
.await?;

let receipt = TestCounter::deploy_builder(provider.clone())
.send()
.await?
.get_receipt()
.await?;
let contract = receipt.contract_address.unwrap();
let contract = TestCounter::TestCounterInstance::new(contract, provider.clone());

let contract_deployment_block = provider
.get_block(receipt.block_number.unwrap().into())
.await?
.expect("block should exist");

// initial event emissions
let event_count = 5;
for _ in 0..event_count {
let _ = contract.increase().send().await?.get_receipt().await?;
}

let latest_block = provider.get_block_number().await?;

// assert the total number of logs in the whole chain
let log_filter = &Filter::new().from_block(0);

let logs = provider.get_logs(log_filter).await?;
assert_eq!(event_count, logs.len() as u64);

// reorg all the blocks that include contract logs, re-emitting only 3 events
let reorg_depth = latest_block - contract_deployment_block.number();
let new_event_count = 3;

let tx_block_pairs = (0..new_event_count)
.map(|_| {
let tx = contract.increase().into_transaction_request();
(TransactionData::JSON(tx), 0)
})
.collect();

let opts = ReorgOptions {
depth: reorg_depth,
tx_block_pairs,
};
provider.anvil_reorg(opts).await?;

// after the reorg, the block when contract was deployed should be unchanged
let post_reorg_contract_deployment_block = provider
.get_block(contract_deployment_block.number().into())
.await?
.expect("block should exist");
assert_eq!(
contract_deployment_block.hash(),
post_reorg_contract_deployment_block.hash()
);
assert_eq!(1, post_reorg_contract_deployment_block.transactions.len());

// the next block should contain all of the new transactions
let txs_block = contract_deployment_block.number() + 1;
let txs_block = provider
.get_block(txs_block.into())
.await?
.expect("block should exist");
assert_eq!(new_event_count, txs_block.transactions.len());

// other blocks should contain no transactions
for i in (txs_block.number() + 1)..latest_block {
let block = provider
.get_block(i.into())
.await?
.expect("block should exist");
assert!(block.transactions.is_empty());
}

// reassert the number of logs in the whole chain
let logs = provider.get_logs(log_filter).await?;
assert_eq!(new_event_count, logs.len()); // FAIL: logs.len() somehow equals 0 (zero)

Ok(())
}
}
}
```
7. Open terminal in the root of the project
8. Run `cargo test`
9. Result:
- **Expected:** the test passes
- **Actual:** the test _fails_ with:
```
running 4 tests
test tests::block_per_tx::test_reorg_no_initial_logs ... ok
test tests::block_per_tx::test_reorg_all_log_emissions ... FAILED
test tests::block_per_tx::test_reorg_all_log_emissions_but_first ... ok
test tests::with_block_time::test_reorg_all_log_emissions ... FAILED

failures:

---- tests::block_per_tx::test_reorg_all_log_emissions stdout ----

thread 'tests::block_per_tx::test_reorg_all_log_emissions' (214686) panicked at src/lib.rs:130:13:
assertion `left == right` failed
left: 3
right: 0
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

---- tests::with_block_time::test_reorg_all_log_emissions stdout ----

thread 'tests::with_block_time::test_reorg_all_log_emissions' (214689) panicked at src/lib.rs:393:13:
assertion `left == right` failed
left: 3
right: 0

failures:
tests::block_per_tx::test_reorg_all_log_emissions
tests::with_block_time::test_reorg_all_log_emissions
```

For faster repro, I created a repository with the above setup, so you can just clone the repo and repro the bug on your machine:
https://github.com/0xNeshi/alloy-anvil-reorg-bug

_NOTE: also tested this with alloy@1.0.30 and alloy-node-bindings@1.0.30._

---

Potentially related issues:
- https://github.com/alloy-rs/alloy/issues/2668
- https://github.com/foundry-rs/foundry/issues/10809
- https://github.com/foundry-rs/foundry/issues/9933