diff --git a/CHANGELOG.md b/CHANGELOG.md index 330b1f3..ad83f10 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -137,8 +137,16 @@ pre-release development phases (see [`docs/ROADMAP.md`](docs/ROADMAP.md)). `ReactiveRegistry` exposes consolidated Alloy log filters for provider subscription setup and exact local log routing with optional route keys. Includes a provider-agnostic `EventSubscriber` trait, an `AlloySubscriber` - scaffold for future live transport work, and an adapter from legacy - `EventDecoder`s to reactive handlers. Generic core. + that drives default WebSocket/pubsub Alloy `subscribe_logs`, + `subscribe_blocks`, and `subscribe_pending_transactions` streams for live log, + block-header, and pending transaction hash inputs, with automatic + source-specific reconnect that retries immediately first, attempts three + times by default, and performs log gap backfill from the last seen block when + a pubsub stream terminates, plus an adapter from legacy `EventDecoder`s to + reactive handlers. HTTP polling `watch_logs` / `watch_pending_transactions` + support is still exported behind the opt-in `reactive-polling` feature. Full + block bodies, full pending transaction hydration, and arbitrary historical + backfill remain follow-up transport work. Generic core. - **Reactive storage resync execution** — `ReactiveRuntime::ingest_batch_with_resync` preserves the direct-effect behavior of `ingest_batch`, then executes surfaced storage resync requests through `EvmCache`'s provider-neutral diff --git a/Cargo.lock b/Cargo.lock index dc8eaf5..7010068 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -364,12 +364,14 @@ dependencies = [ "alloy-network", "alloy-network-primitives", "alloy-primitives", + "alloy-pubsub", "alloy-rpc-client", "alloy-rpc-types-eth", "alloy-signer", "alloy-sol-types", "alloy-transport", "alloy-transport-http", + "alloy-transport-ws", "async-stream", "async-trait", "auto_impl", @@ -390,6 +392,28 @@ dependencies = [ "wasmtimer", ] +[[package]] +name = "alloy-pubsub" +version = "1.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6eebf54983d4fccea08053c218ee5c288adf2e660095a243d0532a8070b43955" +dependencies = [ + "alloy-json-rpc", + "alloy-primitives", + "alloy-transport", + "auto_impl", + "bimap", + "futures", + "parking_lot", + "serde", + "serde_json", + "tokio", + "tokio-stream", + "tower", + "tracing", + "wasmtimer", +] + [[package]] name = "alloy-rlp" version = "0.3.15" @@ -420,8 +444,10 @@ checksum = "91577235d341a1bdbee30a463655d08504408a4d51e9f72edbfc5a622829f402" dependencies = [ "alloy-json-rpc", "alloy-primitives", + "alloy-pubsub", "alloy-transport", "alloy-transport-http", + "alloy-transport-ws", "futures", "pin-project", "reqwest", @@ -652,6 +678,23 @@ dependencies = [ "url", ] +[[package]] +name = "alloy-transport-ws" +version = "1.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5ed38ea573c6658e0c2745af9d1f1773b1ed83aa59fbd9c286358ad469c3233a" +dependencies = [ + "alloy-pubsub", + "alloy-transport", + "futures", + "http", + "serde_json", + "tokio", + "tokio-tungstenite", + "tracing", + "ws_stream_wasm", +] + [[package]] name = "alloy-trie" version = "0.9.5" @@ -1042,6 +1085,17 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "async_io_stream" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6d7b9decdf35d8908a7e3ef02f64c5e9b1695e230154c0e8de3969142d9b94c" +dependencies = [ + "futures", + "pharos", + "rustc_version 0.4.1", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -1093,6 +1147,12 @@ version = "1.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" +[[package]] +name = "bimap" +version = "0.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "230c5f1ca6a325a32553f8640d31ac9b49f2411e901e427570154868b46da4f7" + [[package]] name = "bincode" version = "1.3.3" @@ -1631,6 +1691,12 @@ dependencies = [ "parking_lot_core", ] +[[package]] +name = "data-encoding" +version = "2.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8" + [[package]] name = "der" version = "0.7.10" @@ -1864,6 +1930,7 @@ dependencies = [ "foundry-fork-db", "futures", "revm", + "rustls", "serde", "serde_json", "thiserror", @@ -3087,6 +3154,16 @@ dependencies = [ "ucd-trie", ] +[[package]] +name = "pharos" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e9567389417feee6ce15dd6527a8a1ecac205ef62c2932bcf3d9f6fc5b78b414" +dependencies = [ + "futures", + "rustc_version 0.4.1", +] + [[package]] name = "phf" version = "0.13.1" @@ -3857,6 +3934,20 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "rustls" +version = "0.23.40" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef86cd5876211988985292b91c96a8f2d298df24e75989a43a3c73f2d4d8168b" +dependencies = [ + "once_cell", + "ring", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + [[package]] name = "rustls-pki-types" version = "1.14.1" @@ -3866,6 +3957,17 @@ dependencies = [ "zeroize", ] +[[package]] +name = "rustls-webpki" +version = "0.103.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustversion" version = "1.0.22" @@ -4041,6 +4143,12 @@ dependencies = [ "pest", ] +[[package]] +name = "send_wrapper" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd0b0ec5f1c1ca621c432a25813d8d60c88abe6d3e08a3eb9cf37d97a0fe3d73" + [[package]] name = "serde" version = "1.0.228" @@ -4139,6 +4247,17 @@ dependencies = [ "serde", ] +[[package]] +name = "sha1" +version = "0.10.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "digest 0.10.7", +] + [[package]] name = "sha2" version = "0.10.9" @@ -4505,6 +4624,16 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-rustls" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" +dependencies = [ + "rustls", + "tokio", +] + [[package]] name = "tokio-stream" version = "0.1.18" @@ -4517,6 +4646,22 @@ dependencies = [ "tokio-util", ] +[[package]] +name = "tokio-tungstenite" +version = "0.26.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7a9daff607c6d2bf6c16fd681ccb7eecc83e4e2cdc1ca067ffaadfca5de7f084" +dependencies = [ + "futures-util", + "log", + "rustls", + "rustls-pki-types", + "tokio", + "tokio-rustls", + "tungstenite", + "webpki-roots 0.26.11", +] + [[package]] name = "tokio-util" version = "0.7.18" @@ -4690,6 +4835,25 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.26.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4793cb5e56680ecbb1d843515b23b6de9a75eb04b66643e256a396d43be33c13" +dependencies = [ + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand 0.9.4", + "rustls", + "rustls-pki-types", + "sha1", + "thiserror", + "utf-8", +] + [[package]] name = "typenum" version = "1.20.1" @@ -4757,6 +4921,12 @@ dependencies = [ "serde_derive", ] +[[package]] +name = "utf-8" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" + [[package]] name = "utf8_iter" version = "1.0.4" @@ -4956,6 +5126,24 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "webpki-roots" +version = "0.26.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" +dependencies = [ + "webpki-roots 1.0.8", +] + +[[package]] +name = "webpki-roots" +version = "1.0.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf85cb06032201fa7c6f829d7db5a7e5aa45bcc0655327713065f6f0576731bf" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "winapi-util" version = "0.1.11" @@ -5215,6 +5403,25 @@ version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" +[[package]] +name = "ws_stream_wasm" +version = "0.7.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c173014acad22e83f16403ee360115b38846fe754e735c5d9d3803fe70c6abc" +dependencies = [ + "async_io_stream", + "futures", + "js-sys", + "log", + "pharos", + "rustc_version 0.4.1", + "send_wrapper", + "thiserror", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "wyz" version = "0.5.1" diff --git a/Cargo.toml b/Cargo.toml index 307864b..f491eca 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,8 +16,10 @@ documentation = "https://docs.rs/evm-fork-cache" [workspace] [features] -default = ["reactive"] +default = ["reactive", "reactive-ws"] reactive = [] +reactive-ws = ["reactive", "alloy-provider/ws", "dep:rustls", "rustls/ring"] +reactive-polling = ["reactive"] [dependencies] alloy-consensus = "1.1.2" @@ -36,10 +38,11 @@ anyhow = "1.0.98" bincode = "1.3" foundry-fork-db = "0.22" revm = { version = "34.0", features = ["std", "serde", "optional_eip3607", "optional_no_base_fee", "optional_balance_check"] } +rustls = { version = "0.23", default-features = false, optional = true } serde = { version = "1.0.228", features = ["derive"] } thiserror = "2.0" serde_json = "1.0.145" -tokio = { version = "1.48.0", features = ["rt-multi-thread"] } +tokio = { version = "1.48.0", features = ["rt-multi-thread", "time"] } tracing = "0.1.41" [dev-dependencies] @@ -49,7 +52,8 @@ alloy-transport = "1.0.38" alloy-transport-http = "1.0.38" criterion = "0.5" # `macros` powers `#[tokio::main]`/`#[tokio::test]` in the examples and tests. -tokio = { version = "1.48.0", features = ["macros", "rt-multi-thread"] } +# `time` lets live RPC examples bound their subscription windows. +tokio = { version = "1.48.0", features = ["macros", "rt-multi-thread", "time"] } [[bench]] name = "revert_decoding" @@ -79,6 +83,10 @@ harness = false name = "event_pipeline" harness = false +[[example]] +name = "reactive_alloy_amm_live_probe" +required-features = ["reactive"] + # RPC-gated real-contract benchmarks. Skipped (not failed) when RPC_URL is unset, # so `cargo bench` stays offline by default. [[bench]] diff --git a/README.md b/README.md index bc76e7c..6b45bc9 100644 --- a/README.md +++ b/README.md @@ -58,8 +58,8 @@ around three capabilities that target exactly this workload: - **Event-to-state pipeline** — decode logs into `StateUpdate`s, apply them in order, purge touched state on reorg, and reconcile sampled event-derived slots against RPC. The crate ships the generic driver, the ERC-20 `Transfer` decoder, - and in-memory examples; production WebSocket subscription/reorg wiring and - protocol-specific decoders stay with the consumer or companion crates. + and in-memory examples; protocol-specific decoders stay with the consumer or + companion crates. - **Reactive runtime** — register pure handlers for logs, block notifications, and pending transaction signals. Handlers emit `StateUpdate`s, invalidations, resync requests, speculative signals, and hook signals; the runtime routes @@ -69,9 +69,19 @@ around three capabilities that target exactly this workload: provider-neutral storage batch fetcher before dispatching reports to hooks. The `ReactiveRegistry` exposes consolidated Alloy log filters for provider subscription setup and exact local log routing with optional route keys. The - provider-agnostic `EventSubscriber` trait and `AlloySubscriber` scaffold are - included; live Alloy stream driving is intentionally left to a future transport - layer. + provider-agnostic `EventSubscriber` trait and `AlloySubscriber` are included; + the Alloy subscriber uses WebSocket/pubsub `subscribe_logs`, + `subscribe_blocks`, and `subscribe_pending_transactions` by default for live + log, block-header, and pending-transaction-hash inputs. If an established + WebSocket subscription stream terminates, the subscriber recreates that source + immediately, retries three times by default with exponential backoff between + later attempts, and backfills log subscriptions from the last seen block + through `get_logs`, marking recovered records as `InputSource::Backfill` while + suppressing recent duplicate canonical inputs. HTTP polling `watch_logs` / + `watch_pending_transactions` remains available behind the opt-in + `reactive-polling` feature. Full block bodies, full pending transaction + hydration, and arbitrary historical backfill remain explicit follow-up + transport work. - **ERC20 helpers** — balances, allowances, decimals, and controlled balance mutation (including automatic balance-slot discovery) for simulations. - **Transfer-inspector simulation** that reports per-token balance deltas @@ -198,12 +208,20 @@ endpoint (they print instructions and exit if it is unset): | `multicall_batch` | Intermediate | Batch many view calls through Multicall3 in one pass. | | `multicall_with_error_handling` | Intermediate | Batch with `allowFailure`; read partial results when a call reverts. | | `fork_override_balance` | Intermediate | Discover a real token's balance slot and override it. | +| `reactive_alloy_amm_live_probe` | Advanced | Subscribe to live mainnet AMM logs through the WebSocket-backed `AlloySubscriber`. | ```sh cargo run --example revert_decoding RPC_URL=https://eth.llamarpc.com cargo run --example fork_token_balance +WS_RPC_URL=wss://example-mainnet-endpoint cargo run --example reactive_alloy_amm_live_probe ``` +## Feature Flags + +Default features enable the reactive runtime and WebSocket/pubsub subscriber +support (`reactive`, `reactive-ws`). The HTTP polling subscriber is opt-in: +consumers that disable defaults can enable `reactive,reactive-polling`. + ## Foundry artifact etching Use `etch_foundry_artifact` when replacing an existing forked contract while diff --git a/examples/reactive_alloy_amm_live_probe.rs b/examples/reactive_alloy_amm_live_probe.rs new file mode 100644 index 0000000..cb883c1 --- /dev/null +++ b/examples/reactive_alloy_amm_live_probe.rs @@ -0,0 +1,369 @@ +//! Subscribe to live mainnet AMM logs through `AlloySubscriber`. +//! +//! This example is intentionally RPC-gated. It exits successfully when +//! `RPC_URL` is unset, and it fails if a configured run window sees fewer logs +//! than requested. It also preflights the same filters against recent blocks so +//! a live quiet period can be distinguished from an incorrect AMM filter: +//! +//! ```sh +//! WS_RPC_URL=wss://example-mainnet-endpoint \ +//! LIVE_AMM_SECONDS=90 \ +//! LIVE_AMM_MIN_EVENTS=3 \ +//! LIVE_AMM_PREFLIGHT_BLOCKS=50 \ +//! cargo run --example reactive_alloy_amm_live_probe +//! ``` +//! +//! Default builds use Alloy pubsub/WebSocket `subscribe_logs`. Compile with +//! `--no-default-features --features reactive,reactive-polling` and set +//! `LIVE_AMM_TRANSPORT=polling` to exercise the HTTP `watch_logs` fallback. + +use std::{ + collections::BTreeMap, + time::{Duration, Instant}, +}; + +use alloy_primitives::{Address, B256, address, keccak256}; +#[cfg(feature = "reactive-ws")] +use alloy_provider::WsConnect; +use alloy_provider::{Provider, ProviderBuilder}; +use alloy_rpc_types_eth::{Filter, Log}; +use anyhow::{Context, Result, bail}; +use evm_fork_cache::reactive::{ + AlloySubscriber, ChainStatus, EventSubscriber, InputSource, LogInterest, ReactiveInput, + ReactiveInterest, SubscriberConfig, SubscriberMode, +}; +use tokio::time::timeout; + +#[derive(Clone)] +struct AmmEventTarget { + name: &'static str, + address: Address, + topic0: B256, +} + +impl AmmEventTarget { + fn filter(&self) -> Filter { + Filter::new() + .address(self.address) + .event_signature(self.topic0) + } +} + +fn tracked_amm_events() -> Vec { + vec![ + AmmEventTarget { + name: "UniswapV2 WETH/USDC Swap", + address: address!("B4e16d0168e52d35CaCD2c6185b44281Ec28C9Dc"), + topic0: keccak256(b"Swap(address,uint256,uint256,uint256,uint256,address)"), + }, + AmmEventTarget { + name: "UniswapV2 WETH/USDC Sync", + address: address!("B4e16d0168e52d35CaCD2c6185b44281Ec28C9Dc"), + topic0: keccak256(b"Sync(uint112,uint112)"), + }, + AmmEventTarget { + name: "UniswapV3 USDC/WETH 0.05% Swap", + address: address!("88e6A0c2dDD26FEEb64F039a2c41296FcB3f5640"), + topic0: keccak256(b"Swap(address,address,int256,int256,uint160,uint128,int24)"), + }, + AmmEventTarget { + name: "Balancer V2 Vault Swap", + address: address!("BA12222222228d8Ba445958a75a0704d566BF2C8"), + topic0: keccak256(b"Swap(bytes32,address,address,uint256,uint256)"), + }, + ] +} + +#[tokio::main(flavor = "multi_thread")] +async fn main() -> Result<()> { + if std::env::var("LIVE_AMM_TRANSPORT") + .is_ok_and(|transport| transport.eq_ignore_ascii_case("polling")) + { + #[cfg(feature = "reactive-polling")] + { + return run_polling_probe().await; + } + + #[cfg(not(feature = "reactive-polling"))] + { + bail!("LIVE_AMM_TRANSPORT=polling requires the reactive-polling feature"); + } + } + + #[cfg(feature = "reactive-ws")] + { + return run_ws_probe().await; + } + + #[cfg(all(not(feature = "reactive-ws"), feature = "reactive-polling"))] + { + return run_polling_probe().await; + } + + #[cfg(not(any(feature = "reactive-ws", feature = "reactive-polling")))] + { + eprintln!("No AlloySubscriber transport feature is enabled for this example."); + Ok(()) + } +} + +#[cfg(feature = "reactive-ws")] +async fn run_ws_probe() -> Result<()> { + let Ok(rpc_url) = std::env::var("WS_RPC_URL").or_else(|_| std::env::var("RPC_URL")) else { + eprintln!("WS_RPC_URL not set - skipping reactive_alloy_amm_live_probe."); + eprintln!( + " WS_RPC_URL= cargo run --example reactive_alloy_amm_live_probe" + ); + return Ok(()); + }; + + let provider = ProviderBuilder::new() + .connect_ws(WsConnect::new(rpc_url)) + .await + .context("connect websocket provider")?; + run_probe( + provider, + SubscriberMode::Auto, + InputSource::Subscription, + "Alloy subscribe_logs websocket/pubsub", + ) + .await +} + +#[cfg(feature = "reactive-polling")] +async fn run_polling_probe() -> Result<()> { + let Ok(rpc_url) = std::env::var("RPC_URL") else { + eprintln!("RPC_URL not set - skipping reactive_alloy_amm_live_probe polling mode."); + eprintln!( + " LIVE_AMM_TRANSPORT=polling RPC_URL= cargo run --no-default-features --features reactive,reactive-polling --example reactive_alloy_amm_live_probe" + ); + return Ok(()); + }; + + let provider = ProviderBuilder::new().connect_http(rpc_url.parse().context("valid RPC_URL")?); + run_probe( + provider, + SubscriberMode::Polling, + InputSource::Poll, + "Alloy watch_logs HTTP polling", + ) + .await +} + +async fn run_probe

( + provider: P, + mode: SubscriberMode, + expected_source: InputSource, + transport_label: &'static str, +) -> Result<()> +where + P: Provider + Send + Sync, +{ + let run_seconds = env_u64("LIVE_AMM_SECONDS", 90); + let min_events = env_usize("LIVE_AMM_MIN_EVENTS", 3); + let preflight_blocks = env_u64("LIVE_AMM_PREFLIGHT_BLOCKS", 50); + let targets = tracked_amm_events(); + + println!( + "subscribing to {} AMM event filters for {}s; requiring at least {} log(s)", + targets.len(), + run_seconds, + min_events + ); + println!("transport: {transport_label}"); + + if preflight_blocks > 0 { + preflight_recent_logs(&provider, &targets, preflight_blocks).await?; + } + + let mut subscriber = AlloySubscriber::new( + provider, + mode, + SubscriberConfig { + hydrate_pending_transactions: false, + max_batch_size: 256, + ..SubscriberConfig::default() + }, + ); + + let interests = targets + .iter() + .map(|target| { + ReactiveInterest::Logs(LogInterest { + provider_filter: target.filter(), + local_matcher: None, + route_key: None, + }) + }) + .collect::>(); + subscriber.register_interests(&interests)?; + + let started = Instant::now(); + let run_for = Duration::from_secs(run_seconds); + let mut counts = BTreeMap::<&'static str, usize>::new(); + let mut total = 0usize; + let mut removed = 0usize; + + while started.elapsed() < run_for && total < min_events { + let remaining = run_for.saturating_sub(started.elapsed()); + let batch = match timeout(remaining, subscriber.next_batch()).await { + Ok(Ok(Some(batch))) => batch, + Ok(Ok(None)) => break, + Ok(Err(error)) => return Err(error.into()), + Err(_) => break, + }; + + println!("batch: {} record(s)", batch.records().len()); + for record in batch.records() { + if record.context.source != expected_source { + bail!( + "expected {:?} source, got {:?}", + expected_source, + record.context.source + ); + } + + let ReactiveInput::Log(log) = &record.input else { + bail!("expected only log inputs, got {:?}", record.input); + }; + + let Some(target) = identify_target(&targets, log) else { + bail!( + "subscriber emitted an untracked log from {} topic0={:?}", + log.address(), + log.topics().first() + ); + }; + + match (&record.context.chain_status, log.removed) { + ( + ChainStatus::Included { + block, + confirmations, + }, + false, + ) => { + if Some(block.number) != log.block_number { + bail!( + "context block {} disagrees with log block {:?}", + block.number, + log.block_number + ); + } + if *confirmations != 0 { + bail!("subscriber logs should currently report zero confirmations"); + } + } + (ChainStatus::Reorged { dropped_from }, true) => { + if Some(dropped_from.number) != log.block_number { + bail!( + "reorg context block {} disagrees with removed log block {:?}", + dropped_from.number, + log.block_number + ); + } + removed += 1; + } + (status, was_removed) => { + bail!( + "unexpected chain status {:?} for removed={}", + status, + was_removed + ); + } + } + + *counts.entry(target.name).or_default() += 1; + total += 1; + + println!( + " {:<34} block={} tx={} log_index={:?} removed={}", + target.name, + log.block_number + .map_or_else(|| "pending".to_owned(), |number| number.to_string()), + log.transaction_hash + .map_or_else(|| "-".to_owned(), |hash| hash.to_string()), + log.log_index, + log.removed + ); + } + } + + println!( + "summary: observed {} log(s), {} removed/reorged", + total, removed + ); + for target in &targets { + println!( + " {:<34} {}", + target.name, + counts.get(target.name).copied().unwrap_or_default() + ); + } + + if total < min_events { + bail!( + "observed {} log(s), below LIVE_AMM_MIN_EVENTS={}; increase LIVE_AMM_SECONDS or use a filter-capable RPC endpoint", + total, + min_events + ); + } + + Ok(()) +} + +async fn preflight_recent_logs

( + provider: &P, + targets: &[AmmEventTarget], + blocks: u64, +) -> Result<()> +where + P: Provider, +{ + let latest = provider.get_block_number().await?; + let from = latest.saturating_sub(blocks); + let mut total = 0usize; + + println!( + "preflight: scanning recent AMM logs over blocks {}..={} with the same filters", + from, latest + ); + for target in targets { + let logs = provider + .get_logs(&target.filter().from_block(from).to_block(latest)) + .await?; + total += logs.len(); + println!(" {:<34} {}", target.name, logs.len()); + } + + if total == 0 { + bail!( + "preflight observed zero AMM logs over the last {} block(s); filters or endpoint are not suitable for this probe", + blocks + ); + } + + println!("preflight: observed {} recent log(s)", total); + Ok(()) +} + +fn identify_target<'a>(targets: &'a [AmmEventTarget], log: &Log) -> Option<&'a AmmEventTarget> { + let topic0 = log.topics().first().copied()?; + targets + .iter() + .find(|target| target.address == log.address() && target.topic0 == topic0) +} + +fn env_u64(name: &str, default: u64) -> u64 { + std::env::var(name) + .ok() + .and_then(|value| value.parse().ok()) + .unwrap_or(default) +} + +fn env_usize(name: &str, default: usize) -> usize { + std::env::var(name) + .ok() + .and_then(|value| value.parse().ok()) + .unwrap_or(default) +} diff --git a/src/reactive/mod.rs b/src/reactive/mod.rs index 7ad9ddd..5aac426 100644 --- a/src/reactive/mod.rs +++ b/src/reactive/mod.rs @@ -14,13 +14,14 @@ use std::{ any::Any, borrow::Cow, - collections::{HashMap, HashSet}, + collections::{HashMap, HashSet, VecDeque}, fmt, future::Future, hash::Hash, marker::PhantomData, pin::Pin, sync::Arc, + time::Duration, }; use alloy_consensus::{BlockHeader as _, Transaction as _}; @@ -28,11 +29,17 @@ use alloy_eips::BlockId; use alloy_network::{ Ethereum, Network, primitives::{ - BlockResponse as _, HeaderResponse as _, TransactionResponse as TransactionResponseTrait, + BlockResponse as _, HeaderResponse as HeaderResponseTrait, + TransactionResponse as TransactionResponseTrait, }, }; use alloy_primitives::{Address, B256, Bytes, U256}; +use alloy_provider::Provider; use alloy_rpc_types_eth::{Filter, FilterSet, Log}; +use futures::{ + StreamExt, + stream::{self, BoxStream, SelectAll}, +}; use crate::{ cache::EvmCache, @@ -2037,13 +2044,19 @@ pub type SubscriberNextBatch<'a, N> = Pin< Box>, SubscriberError>> + Send + 'a>, >; -/// Subscriber mode requested for the Alloy scaffold. +/// Subscriber mode requested for the Alloy subscriber. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)] pub enum SubscriberMode { - /// Use provider pubsub streams when available. + /// Prefer the default compiled transport. + /// + /// With the default `reactive-ws` feature this resolves to pubsub/WebSocket + /// subscriptions. Without `reactive-ws`, it resolves to polling only when + /// the opt-in `reactive-polling` feature is enabled. #[default] + Auto, + /// Use provider pubsub streams. PubSub, - /// Use polling/watch APIs. + /// Use polling/watch APIs. Requires the `reactive-polling` feature. Polling, } @@ -2054,6 +2067,8 @@ pub struct SubscriberConfig { pub hydrate_pending_transactions: bool, /// Maximum records to emit per batch. pub max_batch_size: usize, + /// Reconnect policy for WebSocket/pubsub streams. + pub reconnect: SubscriberReconnectConfig, } impl Default for SubscriberConfig { @@ -2061,31 +2076,84 @@ impl Default for SubscriberConfig { Self { hydrate_pending_transactions: false, max_batch_size: 1024, + reconnect: SubscriberReconnectConfig::default(), + } + } +} + +/// WebSocket/pubsub reconnect policy. +/// +/// Reconnects are applied after an established subscription stream terminates. +/// Initial subscription failures are still returned immediately so deployment +/// mistakes, unsupported transports, and bad endpoints fail fast. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SubscriberReconnectConfig { + /// Whether pubsub streams should be recreated after termination. + pub enabled: bool, + /// Delay before the first reconnect attempt. + pub initial_delay: Duration, + /// Delay before the second reconnect attempt. Later retries double this + /// delay up to [`Self::max_delay`]. + pub retry_delay: Duration, + /// Maximum delay between reconnect attempts. + pub max_delay: Duration, + /// Maximum reconnect attempts per terminated stream. `None` retries forever. + pub max_attempts: Option, + /// Number of recently emitted canonical input refs remembered to suppress + /// duplicates across reconnect backfill and subscription replay. + pub dedupe_window: usize, +} + +impl Default for SubscriberReconnectConfig { + fn default() -> Self { + Self { + enabled: true, + initial_delay: Duration::ZERO, + retry_delay: Duration::from_millis(250), + max_delay: Duration::from_secs(30), + max_attempts: Some(3), + dedupe_window: 4096, } } } -/// Documented Alloy subscriber scaffold. +/// Alloy-backed event subscriber. /// -/// This type records interests and configuration but does not yet drive live -/// provider streams. Use it as the integration point for a future -/// `reactive-alloy` transport implementation. +/// The default transport slice drives Alloy pubsub subscriptions for logs, +/// block headers, and pending transaction hashes. The HTTP polling `watch_*` +/// transport remains available behind the opt-in `reactive-polling` feature. +/// Pubsub streams reconnect automatically after termination, and log +/// subscriptions are backfilled from the last seen block. Full pending +/// transaction hydration, full block bodies, and arbitrary historical backfill +/// remain explicit follow-up work. +/// With no registered interests, [`EventSubscriber::next_batch`] returns +/// `Ok(None)`. pub struct AlloySubscriber { provider: P, mode: SubscriberMode, config: SubscriberConfig, interests: Vec>, + state: AlloySubscriberState, + pending_records: VecDeque>, + last_seen_log_blocks: HashMap, + recent_input_refs: VecDeque, + recent_input_ref_set: HashSet, _network: PhantomData, } impl AlloySubscriber { - /// Create a new Alloy subscriber scaffold. + /// Create a new Alloy subscriber. pub fn new(provider: P, mode: SubscriberMode, config: SubscriberConfig) -> Self { Self { provider, mode, config, interests: Vec::new(), + state: AlloySubscriberState::Uninitialized, + pending_records: VecDeque::new(), + last_seen_log_blocks: HashMap::new(), + recent_input_refs: VecDeque::new(), + recent_input_ref_set: HashSet::new(), _network: PhantomData, } } @@ -2109,29 +2177,1035 @@ impl AlloySubscriber { pub fn registered_interests(&self) -> &[ReactiveInterest] { &self.interests } + + fn drain_next_batch(&mut self) -> Option> { + if self.pending_records.is_empty() { + return None; + } + + let len = self.config.max_batch_size.min(self.pending_records.len()); + let records = self.pending_records.drain(..len).collect(); + Some(ReactiveInputBatch::new(records)) + } + + fn reset_delivery_state(&mut self) { + self.pending_records.clear(); + self.last_seen_log_blocks.clear(); + self.recent_input_refs.clear(); + self.recent_input_ref_set.clear(); + } +} + +enum AlloySubscriberState { + Uninitialized, + Active(SubscriberStreams), + Empty, +} + +struct SubscriberStreams { + streams: SelectAll>>, +} + +impl SubscriberStreams { + fn new() -> Self { + Self { + streams: SelectAll::new(), + } + } + + fn is_empty(&self) -> bool { + self.streams.is_empty() + } + + fn push(&mut self, stream: BoxStream<'static, SubscriberEvent>) { + self.streams.push(stream); + } + + async fn next(&mut self) -> Option> { + self.streams.next().await + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[allow(dead_code)] +enum SubscriberTransport { + PubSub, + Polling, +} + +#[derive(Clone, Debug)] +enum SubscriberStreamSource { + PubSubLog { id: usize, filter: Filter }, + PubSubPendingHashes, + PubSubBlockHeaders, + PollingLog { filter: Filter }, + PollingPendingHashes, +} + +impl SubscriberStreamSource { + fn label(&self) -> &'static str { + match self { + Self::PubSubLog { .. } => "pubsub log", + Self::PubSubPendingHashes => "pubsub pending transaction hash", + Self::PubSubBlockHeaders => "pubsub block header", + Self::PollingLog { .. } => "polling log", + Self::PollingPendingHashes => "polling pending transaction hash", + } + } + + fn is_pubsub(&self) -> bool { + matches!( + self, + Self::PubSubLog { .. } | Self::PubSubPendingHashes | Self::PubSubBlockHeaders + ) + } } -impl EventSubscriber for AlloySubscriber { +#[allow(dead_code)] +enum SubscriberEvent { + Log { source_id: usize, log: Log }, + BackfilledLogs { source_id: usize, logs: Vec }, + Logs(Vec), + BlockHeader(N::HeaderResponse), + PendingHash(B256), + PendingHashes(Vec), + StreamTerminated(SubscriberStreamSource), +} + +impl EventSubscriber for AlloySubscriber +where + P: Provider + Send + Sync, + N: Network + 'static, + N::HeaderResponse: Send + 'static, +{ fn register_interests( &mut self, interests: &[ReactiveInterest], ) -> Result<(), SubscriberError> { + validate_subscriber_config(&self.config)?; + validate_supported_interests(self.mode, &self.config, interests)?; + self.interests = interests.to_vec(); + self.reset_delivery_state(); + self.state = AlloySubscriberState::Uninitialized; Ok(()) } fn next_batch(&mut self) -> SubscriberNextBatch<'_, N> { Box::pin(async { + if let Some(batch) = self.drain_next_batch() { + return Ok(Some(batch)); + } + + if self.interests.is_empty() { + return Ok(None); + } + + if matches!(self.state, AlloySubscriberState::Uninitialized) { + let streams = self.init_streams().await?; + self.state = if streams.is_empty() { + AlloySubscriberState::Empty + } else { + AlloySubscriberState::Active(streams) + }; + } + + loop { + let Some(event) = self.next_event().await? else { + return Ok(None); + }; + + self.enqueue_event(event); + if let Some(batch) = self.drain_next_batch() { + return Ok(Some(batch)); + } + } + }) + } +} + +impl AlloySubscriber +where + P: Provider + Send + Sync, + N: Network + 'static, + N::HeaderResponse: Send + 'static, +{ + async fn init_streams(&mut self) -> Result, SubscriberError> { + let mut streams = SubscriberStreams::new(); + for source in self.stream_sources()? { + streams.push(self.connect_source_stream(source).await?); + } + Ok(streams) + } + + fn stream_sources(&self) -> Result, SubscriberError> { + match resolve_subscriber_transport(self.mode)? { + SubscriberTransport::PubSub => Ok(pubsub_stream_sources(&self.interests)), + SubscriberTransport::Polling => Ok(polling_stream_sources(&self.interests)), + } + } + + async fn connect_source_stream( + &mut self, + source: SubscriberStreamSource, + ) -> Result>, SubscriberError> { + match source { + SubscriberStreamSource::PubSubLog { id, filter } => { + self.connect_pubsub_log_stream(id, filter).await + } + SubscriberStreamSource::PubSubPendingHashes => { + self.connect_pubsub_pending_hash_stream().await + } + SubscriberStreamSource::PubSubBlockHeaders => { + self.connect_pubsub_block_header_stream().await + } + SubscriberStreamSource::PollingLog { filter } => { + self.connect_polling_log_stream(filter).await + } + SubscriberStreamSource::PollingPendingHashes => { + self.connect_polling_pending_hash_stream().await + } + } + } + + async fn connect_pubsub_log_stream( + &mut self, + id: usize, + filter: Filter, + ) -> Result>, SubscriberError> { + #[cfg(feature = "reactive-ws")] + { + let source = SubscriberStreamSource::PubSubLog { + id, + filter: filter.clone(), + }; + let stream = self + .provider + .subscribe_logs(&filter) + .channel_size(self.config.max_batch_size.max(1)) + .await + .map_err(provider_error)? + .into_stream() + .map(move |log| SubscriberEvent::Log { source_id: id, log }); + Ok(stream_with_termination(stream, source)) + } + + #[cfg(not(feature = "reactive-ws"))] + { + let _ = (id, filter); Err(SubscriberError::Unsupported( - "AlloySubscriber is a scaffold; live stream driving is not implemented in this feature slice", + "AlloySubscriber pubsub mode requires the reactive-ws feature", )) - }) + } + } + + async fn connect_pubsub_pending_hash_stream( + &mut self, + ) -> Result>, SubscriberError> { + #[cfg(feature = "reactive-ws")] + { + let stream = self + .provider + .subscribe_pending_transactions() + .channel_size(self.config.max_batch_size.max(1)) + .await + .map_err(provider_error)? + .into_stream() + .map(SubscriberEvent::PendingHash); + Ok(stream_with_termination( + stream, + SubscriberStreamSource::PubSubPendingHashes, + )) + } + + #[cfg(not(feature = "reactive-ws"))] + { + Err(SubscriberError::Unsupported( + "AlloySubscriber pubsub mode requires the reactive-ws feature", + )) + } + } + + async fn connect_pubsub_block_header_stream( + &mut self, + ) -> Result>, SubscriberError> { + #[cfg(feature = "reactive-ws")] + { + let stream = self + .provider + .subscribe_blocks() + .channel_size(self.config.max_batch_size.max(1)) + .await + .map_err(provider_error)? + .into_stream() + .map(SubscriberEvent::BlockHeader); + Ok(stream_with_termination( + stream, + SubscriberStreamSource::PubSubBlockHeaders, + )) + } + + #[cfg(not(feature = "reactive-ws"))] + { + Err(SubscriberError::Unsupported( + "AlloySubscriber pubsub mode requires the reactive-ws feature", + )) + } + } + + async fn connect_polling_log_stream( + &mut self, + filter: Filter, + ) -> Result>, SubscriberError> { + #[cfg(feature = "reactive-polling")] + { + let source = SubscriberStreamSource::PollingLog { + filter: filter.clone(), + }; + let stream = self + .provider + .watch_logs(&filter) + .await + .map_err(provider_error)? + .with_channel_size(self.config.max_batch_size.max(1)) + .into_stream() + .map(SubscriberEvent::Logs); + Ok(stream_with_termination(stream, source)) + } + + #[cfg(not(feature = "reactive-polling"))] + { + let _ = filter; + Err(SubscriberError::Unsupported( + "AlloySubscriber polling mode requires the reactive-polling feature", + )) + } + } + + async fn connect_polling_pending_hash_stream( + &mut self, + ) -> Result>, SubscriberError> { + #[cfg(feature = "reactive-polling")] + { + let stream = self + .provider + .watch_pending_transactions() + .await + .map_err(provider_error)? + .with_channel_size(self.config.max_batch_size.max(1)) + .into_stream() + .map(SubscriberEvent::PendingHashes); + Ok(stream_with_termination( + stream, + SubscriberStreamSource::PollingPendingHashes, + )) + } + + #[cfg(not(feature = "reactive-polling"))] + { + Err(SubscriberError::Unsupported( + "AlloySubscriber polling mode requires the reactive-polling feature", + )) + } + } + + async fn next_event(&mut self) -> Result>, SubscriberError> { + loop { + let event = match &mut self.state { + AlloySubscriberState::Active(streams) => streams.next().await, + AlloySubscriberState::Uninitialized | AlloySubscriberState::Empty => { + return Ok(None); + } + }; + + let Some(event) = event else { + return Err(SubscriberError::Provider( + "Alloy subscriber streams terminated before the subscriber was stopped" + .to_owned(), + )); + }; + + match event { + SubscriberEvent::StreamTerminated(source) => { + if let Some(backfill_event) = self.reconnect_source_stream(source).await? { + return Ok(Some(backfill_event)); + } + } + event => return Ok(Some(event)), + } + } + } + + fn enqueue_event(&mut self, event: SubscriberEvent) { + match event { + SubscriberEvent::Log { source_id, log } => { + if log_matches_any_interest(&log, &self.interests) { + let record = log_input_record(log, InputSource::Subscription); + self.note_log_block(source_id, &record); + self.enqueue_record(record); + } + } + SubscriberEvent::BackfilledLogs { source_id, logs } => { + for log in logs { + if log_matches_any_interest(&log, &self.interests) { + let record = log_input_record(log, InputSource::Backfill); + self.note_log_block(source_id, &record); + self.enqueue_record(record); + } + } + } + SubscriberEvent::Logs(logs) => self.pending_records.extend( + logs.into_iter() + .filter(|log| log_matches_any_interest(log, &self.interests)) + .map(|log| log_input_record(log, InputSource::Poll)), + ), + SubscriberEvent::BlockHeader(header) => { + if needs_header_block_stream(&self.interests) { + let record = block_header_input_record::(header); + self.enqueue_record(record); + } + } + SubscriberEvent::PendingHash(hash) => { + let record = pending_hash_input_record::(hash, InputSource::Subscription); + self.enqueue_record(record); + } + SubscriberEvent::PendingHashes(hashes) => self.pending_records.extend( + hashes + .into_iter() + .map(|hash| pending_hash_input_record::(hash, InputSource::Poll)), + ), + SubscriberEvent::StreamTerminated(_) => {} + } + } + + async fn reconnect_source_stream( + &mut self, + source: SubscriberStreamSource, + ) -> Result>, SubscriberError> { + if !source.is_pubsub() { + return Err(stream_terminated_error(&source)); + } + + if !self.config.reconnect.enabled { + return Err(SubscriberError::Provider(format!( + "Alloy subscriber {} stream terminated and reconnect is disabled", + source.label() + ))); + } + + let mut attempts = 0usize; + let mut delay = self.config.reconnect.initial_delay; + let mut retry_delay = self.config.reconnect.retry_delay; + + loop { + attempts = attempts.saturating_add(1); + if !delay.is_zero() { + tokio::time::sleep(delay).await; + } + + match self.reconnect_source_once(source.clone()).await { + Ok(backfill_event) => return Ok(backfill_event), + Err(error) if reconnect_attempts_exhausted(attempts, &self.config.reconnect) => { + return Err(SubscriberError::Provider(format!( + "Alloy subscriber {} stream terminated and reconnect failed after {attempts} attempt(s): {error}", + source.label() + ))); + } + Err(error) => { + tracing::warn!( + stream = source.label(), + attempts, + error = %error, + "Alloy subscriber reconnect attempt failed" + ); + delay = retry_delay; + retry_delay = + next_reconnect_delay(retry_delay, self.config.reconnect.max_delay); + } + } + } + } + + async fn reconnect_source_once( + &mut self, + source: SubscriberStreamSource, + ) -> Result>, SubscriberError> { + let stream = self.connect_source_stream(source.clone()).await?; + let backfill_event = self.backfill_reconnected_source(&source).await?; + + match &mut self.state { + AlloySubscriberState::Active(streams) => streams.push(stream), + AlloySubscriberState::Uninitialized | AlloySubscriberState::Empty => { + return Err(SubscriberError::Provider( + "Alloy subscriber state changed before reconnect completed".to_owned(), + )); + } + } + + Ok(backfill_event) } + + async fn backfill_reconnected_source( + &mut self, + source: &SubscriberStreamSource, + ) -> Result>, SubscriberError> { + let SubscriberStreamSource::PubSubLog { id, filter } = source else { + return Ok(None); + }; + let Some(from_block) = self.last_seen_log_blocks.get(id).copied() else { + return Ok(None); + }; + + let latest = self + .provider + .get_block_number() + .await + .map_err(provider_error)?; + if latest < from_block { + return Ok(None); + } + + let logs = self + .provider + .get_logs(&filter.clone().from_block(from_block).to_block(latest)) + .await + .map_err(provider_error)?; + Ok(Some(SubscriberEvent::BackfilledLogs { + source_id: *id, + logs, + })) + } + + fn note_log_block(&mut self, source_id: usize, record: &ReactiveInputRecord) { + if let Some(block) = record.context.block.as_ref() { + self.last_seen_log_blocks.insert(source_id, block.number); + } + } + + fn enqueue_record(&mut self, record: ReactiveInputRecord) { + if self.should_skip_recent_duplicate(&record) { + return; + } + self.remember_record(&record); + self.pending_records.push_back(record); + } + + fn should_skip_recent_duplicate(&self, record: &ReactiveInputRecord) -> bool { + if !should_dedupe_record(record) { + return false; + } + self.recent_input_ref_set.contains(&record.input_ref()) + } + + fn remember_record(&mut self, record: &ReactiveInputRecord) { + if !should_dedupe_record(record) || self.config.reconnect.dedupe_window == 0 { + return; + } + + let input_ref = record.input_ref(); + if !self.recent_input_ref_set.insert(input_ref) { + return; + } + self.recent_input_refs.push_back(input_ref); + + while self.recent_input_refs.len() > self.config.reconnect.dedupe_window { + if let Some(evicted) = self.recent_input_refs.pop_front() { + self.recent_input_ref_set.remove(&evicted); + } + } + } +} + +fn stream_with_termination( + stream: S, + source: SubscriberStreamSource, +) -> BoxStream<'static, SubscriberEvent> +where + N: Network + 'static, + S: futures::Stream> + Send + 'static, +{ + stream + .chain(stream::once(async move { + SubscriberEvent::StreamTerminated(source) + })) + .boxed() +} + +fn pubsub_stream_sources( + interests: &[ReactiveInterest], +) -> Vec { + let mut sources = Vec::new(); + + for (id, filter) in log_filters(interests).into_iter().enumerate() { + sources.push(SubscriberStreamSource::PubSubLog { id, filter }); + } + + if needs_pending_hash_stream(interests) { + sources.push(SubscriberStreamSource::PubSubPendingHashes); + } + + if needs_header_block_stream(interests) { + sources.push(SubscriberStreamSource::PubSubBlockHeaders); + } + + sources +} + +fn polling_stream_sources( + interests: &[ReactiveInterest], +) -> Vec { + let mut sources = Vec::new(); + + for filter in log_filters(interests) { + sources.push(SubscriberStreamSource::PollingLog { filter }); + } + + if needs_pending_hash_stream(interests) { + sources.push(SubscriberStreamSource::PollingPendingHashes); + } + + sources +} + +fn stream_terminated_error(source: &SubscriberStreamSource) -> SubscriberError { + SubscriberError::Provider(format!( + "Alloy subscriber {} stream terminated before the subscriber was stopped", + source.label() + )) +} + +fn reconnect_attempts_exhausted(attempts: usize, config: &SubscriberReconnectConfig) -> bool { + config + .max_attempts + .is_some_and(|max_attempts| attempts >= max_attempts) +} + +fn next_reconnect_delay(current: Duration, max: Duration) -> Duration { + if current.is_zero() { + return current; + } + current.checked_mul(2).unwrap_or(max).min(max) +} + +fn should_dedupe_record(record: &ReactiveInputRecord) -> bool { + match &record.input { + ReactiveInput::Log(log) => { + is_canonical_status(&record.context.chain_status) && !log.removed + } + ReactiveInput::BlockHeader(_) | ReactiveInput::PendingTxHash(_) => true, + ReactiveInput::FullBlock(_) | ReactiveInput::PendingTx(_) => false, + } +} + +#[cfg(test)] +mod subscriber_helper_tests { + use super::*; + use alloy_provider::ProviderBuilder; + use alloy_transport::mock::Asserter; + + fn rpc_log(removed: bool) -> Log { + Log { + inner: alloy_primitives::Log::new_unchecked( + Address::repeat_byte(0x42), + vec![B256::repeat_byte(0x01)], + Bytes::new(), + ), + block_hash: Some(B256::repeat_byte(0x02)), + block_number: Some(7), + block_timestamp: Some(1_700_000_000), + transaction_hash: Some(B256::repeat_byte(0x03)), + transaction_index: Some(4), + log_index: Some(5), + removed, + } + } + + #[tokio::test(flavor = "multi_thread")] + async fn stream_with_termination_yields_terminal_source_marker() { + let mut stream = stream_with_termination::( + stream::iter([SubscriberEvent::::PendingHash(B256::repeat_byte( + 0xaa, + ))]), + SubscriberStreamSource::PubSubPendingHashes, + ); + + assert!(matches!( + stream.next().await, + Some(SubscriberEvent::PendingHash(hash)) if hash == B256::repeat_byte(0xaa) + )); + assert!(matches!( + stream.next().await, + Some(SubscriberEvent::StreamTerminated(source)) if source.is_pubsub() + )); + assert!(stream.next().await.is_none()); + } + + #[test] + fn reconnect_delay_doubles_until_capped() { + assert_eq!( + next_reconnect_delay(Duration::from_millis(250), Duration::from_secs(1)), + Duration::from_millis(500) + ); + assert_eq!( + next_reconnect_delay(Duration::from_millis(750), Duration::from_secs(1)), + Duration::from_secs(1) + ); + assert_eq!( + next_reconnect_delay(Duration::ZERO, Duration::from_secs(1)), + Duration::ZERO + ); + } + + #[test] + fn canonical_logs_are_deduped_but_removed_logs_are_not() { + let included = log_input_record::(rpc_log(false), InputSource::Subscription); + let removed = log_input_record::(rpc_log(true), InputSource::Subscription); + + assert!(should_dedupe_record(&included)); + assert!(!should_dedupe_record(&removed)); + } + + #[test] + fn pubsub_sources_assign_stable_log_ids_before_shared_streams() { + let sources = pubsub_stream_sources::(&[ + ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(Address::repeat_byte(0x01)), + local_matcher: None, + route_key: None, + }), + ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(Address::repeat_byte(0x02)), + local_matcher: None, + route_key: None, + }), + ReactiveInterest::PendingTransactions(PendingTxInterest::default()), + ]); + + assert!(matches!( + &sources[0], + SubscriberStreamSource::PubSubLog { id: 0, .. } + )); + assert!(matches!( + sources[1], + SubscriberStreamSource::PubSubPendingHashes + )); + } + + #[tokio::test(flavor = "multi_thread")] + #[cfg(feature = "reactive-ws")] + async fn pubsub_stream_termination_attempts_reconnect_before_error() { + let provider = ProviderBuilder::new().connect_mocked_client(Asserter::new()); + let mut subscriber = AlloySubscriber::new( + provider, + SubscriberMode::PubSub, + SubscriberConfig { + reconnect: SubscriberReconnectConfig { + initial_delay: Duration::ZERO, + retry_delay: Duration::ZERO, + max_delay: Duration::ZERO, + max_attempts: Some(1), + ..SubscriberReconnectConfig::default() + }, + ..SubscriberConfig::default() + }, + ); + subscriber.interests = vec![ReactiveInterest::PendingTransactions( + PendingTxInterest::default(), + )]; + + let mut streams = SubscriberStreams::new(); + streams.push( + stream::once(async { + SubscriberEvent::::StreamTerminated( + SubscriberStreamSource::PubSubPendingHashes, + ) + }) + .boxed(), + ); + subscriber.state = AlloySubscriberState::Active(streams); + + let result = subscriber.next_batch().await; + assert!( + matches!(result, Err(SubscriberError::Provider(ref message)) if message.contains("reconnect failed after 1 attempt")), + "terminated pubsub streams should attempt reconnect before surfacing failure: {result:?}" + ); + } + + #[test] + fn backfilled_logs_skip_recent_subscription_duplicates() { + let provider = ProviderBuilder::new().connect_mocked_client(Asserter::new()); + let mut subscriber = AlloySubscriber::<_, Ethereum>::new( + provider, + SubscriberMode::PubSub, + SubscriberConfig::default(), + ); + subscriber.interests = vec![ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new() + .address(Address::repeat_byte(0x42)) + .event_signature(B256::repeat_byte(0x01)), + local_matcher: None, + route_key: None, + })]; + + let log = rpc_log(false); + subscriber.enqueue_event(SubscriberEvent::Log { + source_id: 0, + log: log.clone(), + }); + subscriber.enqueue_event(SubscriberEvent::BackfilledLogs { + source_id: 0, + logs: vec![log], + }); + + assert_eq!(subscriber.pending_records.len(), 1); + assert_eq!(subscriber.last_seen_log_blocks.get(&0), Some(&7)); + assert_eq!( + subscriber.pending_records[0].context.source, + InputSource::Subscription + ); + } +} + +fn resolve_subscriber_transport( + mode: SubscriberMode, +) -> Result { + match mode { + SubscriberMode::PubSub => { + #[cfg(feature = "reactive-ws")] + { + Ok(SubscriberTransport::PubSub) + } + #[cfg(not(feature = "reactive-ws"))] + { + Err(SubscriberError::Unsupported( + "AlloySubscriber pubsub mode requires the reactive-ws feature", + )) + } + } + SubscriberMode::Polling => { + #[cfg(feature = "reactive-polling")] + { + Ok(SubscriberTransport::Polling) + } + #[cfg(not(feature = "reactive-polling"))] + { + Err(SubscriberError::Unsupported( + "AlloySubscriber polling mode requires the reactive-polling feature", + )) + } + } + SubscriberMode::Auto => resolve_auto_subscriber_transport(), + } +} + +fn resolve_auto_subscriber_transport() -> Result { + #[cfg(feature = "reactive-ws")] + { + Ok(SubscriberTransport::PubSub) + } + + #[cfg(all(not(feature = "reactive-ws"), feature = "reactive-polling"))] + { + Ok(SubscriberTransport::Polling) + } + + #[cfg(not(any(feature = "reactive-ws", feature = "reactive-polling")))] + { + Err(SubscriberError::Unsupported( + "AlloySubscriber requires either reactive-ws or reactive-polling", + )) + } +} + +fn validate_subscriber_config(config: &SubscriberConfig) -> Result<(), SubscriberError> { + if config.max_batch_size == 0 { + return Err(SubscriberError::InvalidConfig( + "SubscriberConfig::max_batch_size must be greater than zero", + )); + } + if config.reconnect.enabled { + if config.reconnect.retry_delay > config.reconnect.max_delay { + return Err(SubscriberError::InvalidConfig( + "SubscriberReconnectConfig::retry_delay must be less than or equal to max_delay", + )); + } + if matches!(config.reconnect.max_attempts, Some(0)) { + return Err(SubscriberError::InvalidConfig( + "SubscriberReconnectConfig::max_attempts must be greater than zero when set", + )); + } + } + Ok(()) +} + +fn validate_supported_interests( + mode: SubscriberMode, + config: &SubscriberConfig, + interests: &[ReactiveInterest], +) -> Result<(), SubscriberError> { + let transport = resolve_subscriber_transport(mode)?; + + for interest in interests { + match interest { + ReactiveInterest::Logs(_) => {} + ReactiveInterest::PendingTransactions(interest) + if !config.hydrate_pending_transactions && interest.matches_hash_only() => {} + ReactiveInterest::PendingTransactions(_) => { + return Err(SubscriberError::Unsupported( + "AlloySubscriber polling mode currently supports pending transaction hash interests only", + )); + } + ReactiveInterest::Blocks(interest) => match (transport, interest.mode) { + (SubscriberTransport::PubSub, BlockInterestMode::Header) => {} + (_, BlockInterestMode::FullBlock) => { + return Err(SubscriberError::Unsupported( + "AlloySubscriber full block streams are not implemented in this transport slice", + )); + } + (SubscriberTransport::Polling, BlockInterestMode::Header) => { + return Err(SubscriberError::Unsupported( + "AlloySubscriber polling block streams are not implemented in this transport slice", + )); + } + }, + } + } + + Ok(()) +} + +fn log_filters(interests: &[ReactiveInterest]) -> Vec { + let mut filters = Vec::new(); + for interest in interests { + if let ReactiveInterest::Logs(interest) = interest { + merge_log_subscription_filter(&mut filters, &interest.provider_filter); + } + } + filters +} + +fn needs_header_block_stream(interests: &[ReactiveInterest]) -> bool { + interests.iter().any(|interest| { + matches!( + interest, + ReactiveInterest::Blocks(BlockInterest { + mode: BlockInterestMode::Header, + }) + ) + }) +} + +fn needs_pending_hash_stream(interests: &[ReactiveInterest]) -> bool { + interests.iter().any(|interest| { + matches!( + interest, + ReactiveInterest::PendingTransactions(interest) if interest.matches_hash_only() + ) + }) +} + +fn log_matches_any_interest(log: &Log, interests: &[ReactiveInterest]) -> bool { + interests.iter().any(|interest| { + matches!( + interest, + ReactiveInterest::Logs(interest) if interest.matches(log) + ) + }) +} + +fn log_input_record(log: Log, source: InputSource) -> ReactiveInputRecord { + let context = log_reactive_context(&log); + ReactiveInputRecord::new( + ReactiveInput::Log(log), + ReactiveContext { source, ..context }, + ) +} + +fn log_reactive_context(log: &Log) -> ReactiveContext { + let block = match (log.block_hash, log.block_number) { + (Some(hash), Some(number)) => Some(BlockRef { + number, + hash, + parent_hash: None, + timestamp: log.block_timestamp, + }), + _ => None, + }; + + let chain_status = match (&block, log.removed) { + (Some(block), true) => ChainStatus::Reorged { + dropped_from: block.clone(), + }, + (Some(block), false) => ChainStatus::Included { + block: block.clone(), + confirmations: 0, + }, + (None, _) => ChainStatus::Pending, + }; + + ReactiveContext { + chain_id: None, + source: InputSource::Poll, + chain_status, + block, + transaction_index: log.transaction_index, + log_index: log.log_index, + } +} + +fn block_header_input_record(header: N::HeaderResponse) -> ReactiveInputRecord +where + N: Network, +{ + let block = BlockRef { + number: header.number(), + hash: HeaderResponseTrait::hash(&header), + parent_hash: Some(header.parent_hash()), + timestamp: Some(header.timestamp()), + }; + ReactiveInputRecord::new( + ReactiveInput::BlockHeader(header), + ReactiveContext { + chain_id: None, + source: InputSource::Subscription, + chain_status: ChainStatus::Included { + block: block.clone(), + confirmations: 0, + }, + block: Some(block), + transaction_index: None, + log_index: None, + }, + ) +} + +fn pending_hash_input_record( + hash: B256, + source: InputSource, +) -> ReactiveInputRecord { + ReactiveInputRecord::new( + ReactiveInput::PendingTxHash(hash), + ReactiveContext { + chain_id: None, + source, + chain_status: ChainStatus::Pending, + block: None, + transaction_index: None, + log_index: None, + }, + ) +} + +fn provider_error(error: impl fmt::Display) -> SubscriberError { + SubscriberError::Provider(error.to_string()) } /// Subscriber error. #[derive(Debug, thiserror::Error)] pub enum SubscriberError { + /// Invalid subscriber configuration. + #[error("{0}")] + InvalidConfig(&'static str), /// Requested subscriber behavior is not implemented. #[error("{0}")] Unsupported(&'static str), diff --git a/tests/reactive_alloy_subscriber.rs b/tests/reactive_alloy_subscriber.rs new file mode 100644 index 0000000..8148cfb --- /dev/null +++ b/tests/reactive_alloy_subscriber.rs @@ -0,0 +1,465 @@ +//! Manager-authored acceptance tests for the out-of-the-box Alloy subscriber. +//! +//! These tests pin the default WebSocket/pubsub subscriber surface and the +//! opt-in HTTP polling fallback. +#![cfg(feature = "reactive")] + +use std::time::Duration; + +use alloy_network::Ethereum; +use alloy_primitives::{Address, keccak256}; +#[cfg(feature = "reactive-polling")] +use alloy_primitives::{B256, Bytes, Log as PrimitiveLog, U256}; +use alloy_provider::ProviderBuilder; +use alloy_rpc_types_eth::Filter; +#[cfg(feature = "reactive-polling")] +use alloy_rpc_types_eth::Log; +use alloy_transport::mock::Asserter; +use anyhow::Result; +#[cfg(feature = "reactive-polling")] +use anyhow::bail; + +#[cfg(feature = "reactive-ws")] +use evm_fork_cache::reactive::BlockInterestMode; +use evm_fork_cache::reactive::{ + AlloySubscriber, BlockInterest, EventSubscriber, LogInterest, PendingTxInterest, + ReactiveInterest, SubscriberConfig, SubscriberError, SubscriberMode, SubscriberReconnectConfig, +}; +#[cfg(feature = "reactive-polling")] +use evm_fork_cache::reactive::{ChainStatus, InputSource, ReactiveInput}; + +#[cfg(feature = "reactive-polling")] +fn rpc_log(address: Address, topic0: B256, block_number: u64, log_index: u64) -> Log { + Log { + inner: PrimitiveLog::new_unchecked(address, vec![topic0], Bytes::new()), + block_hash: Some(B256::repeat_byte(block_number as u8)), + block_number: Some(block_number), + block_timestamp: Some(1_700_000_000 + block_number), + transaction_hash: Some(B256::repeat_byte(0x20 + log_index as u8)), + transaction_index: Some(log_index), + log_index: Some(log_index), + removed: false, + } +} + +#[cfg(feature = "reactive-polling")] +fn removed_rpc_log(address: Address, topic0: B256, block_number: u64, log_index: u64) -> Log { + let mut log = rpc_log(address, topic0, block_number, log_index); + log.removed = true; + log +} + +#[cfg(feature = "reactive-polling")] +fn polling_subscriber( + asserter: Asserter, + max_batch_size: usize, +) -> AlloySubscriber, Ethereum> { + let provider = ProviderBuilder::new().connect_mocked_client(asserter); + AlloySubscriber::new( + provider, + SubscriberMode::Polling, + SubscriberConfig { + hydrate_pending_transactions: false, + max_batch_size, + ..SubscriberConfig::default() + }, + ) +} + +fn mock_subscriber( + mode: SubscriberMode, +) -> AlloySubscriber, Ethereum> { + let provider = ProviderBuilder::new().connect_mocked_client(Asserter::new()); + AlloySubscriber::new(provider, mode, SubscriberConfig::default()) +} + +#[tokio::test(flavor = "multi_thread")] +#[cfg(feature = "reactive-ws")] +async fn alloy_subscriber_auto_mode_uses_pubsub_by_default() -> Result<()> { + let address = Address::repeat_byte(0xef); + let topic0 = keccak256(b"AutoMode(uint256)"); + + let mut subscriber = mock_subscriber(SubscriberMode::Auto); + subscriber.register_interests(&[ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(address).event_signature(topic0), + local_matcher: None, + route_key: None, + })])?; + + assert_eq!(SubscriberMode::default(), SubscriberMode::Auto); + assert_eq!(subscriber.registered_interests().len(), 1); + + let result = subscriber.next_batch().await; + assert!( + matches!(result, Err(SubscriberError::Provider(_))), + "mock providers are not pubsub transports, so default auto/pubsub mode should report a provider error: {result:?}" + ); + + Ok(()) +} + +#[test] +#[cfg(feature = "reactive-ws")] +fn alloy_subscriber_pubsub_accepts_logs_pending_hashes_and_block_headers() -> Result<()> { + let address = Address::repeat_byte(0xab); + let topic0 = keccak256(b"SubscriberLog(uint256)"); + + let mut subscriber = mock_subscriber(SubscriberMode::PubSub); + subscriber.register_interests(&[ + ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(address).event_signature(topic0), + local_matcher: None, + route_key: None, + }), + ReactiveInterest::PendingTransactions(PendingTxInterest::default()), + ReactiveInterest::Blocks(BlockInterest::default()), + ])?; + + assert_eq!(subscriber.registered_interests().len(), 3); + + Ok(()) +} + +#[test] +#[cfg(feature = "reactive-ws")] +fn alloy_subscriber_pubsub_rejects_full_body_modes() -> Result<()> { + let mut subscriber = mock_subscriber(SubscriberMode::PubSub); + let full_pending = subscriber.register_interests(&[ReactiveInterest::PendingTransactions( + PendingTxInterest { + full_transactions: true, + ..PendingTxInterest::default() + }, + )]); + assert!(matches!(full_pending, Err(SubscriberError::Unsupported(_)))); + + let mut subscriber = mock_subscriber(SubscriberMode::PubSub); + let full_block = subscriber.register_interests(&[ReactiveInterest::Blocks(BlockInterest { + mode: BlockInterestMode::FullBlock, + })]); + assert!(matches!(full_block, Err(SubscriberError::Unsupported(_)))); + + Ok(()) +} + +#[test] +#[cfg(not(feature = "reactive-ws"))] +fn alloy_subscriber_pubsub_requires_ws_feature() -> Result<()> { + let mut subscriber = mock_subscriber(SubscriberMode::PubSub); + let result = subscriber.register_interests(&[ReactiveInterest::PendingTransactions( + PendingTxInterest::default(), + )]); + assert!(matches!(result, Err(SubscriberError::Unsupported(_)))); + + Ok(()) +} + +#[test] +#[cfg(not(feature = "reactive-polling"))] +fn alloy_subscriber_polling_requires_polling_feature() -> Result<()> { + let mut subscriber = mock_subscriber(SubscriberMode::Polling); + let result = subscriber.register_interests(&[ReactiveInterest::PendingTransactions( + PendingTxInterest::default(), + )]); + assert!(matches!(result, Err(SubscriberError::Unsupported(_)))); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +#[cfg(feature = "reactive-polling")] +async fn alloy_subscriber_polling_logs_yield_reactive_records() -> Result<()> { + let asserter = Asserter::new(); + let address = Address::repeat_byte(0xab); + let topic0 = keccak256(b"SubscriberLog(uint256)"); + let log = rpc_log(address, topic0, 42, 7); + + asserter.push_success(&U256::from(1)); + asserter.push_success(&vec![log.clone()]); + + let mut subscriber = polling_subscriber(asserter, 16); + subscriber.register_interests(&[ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(address).event_signature(topic0), + local_matcher: None, + route_key: None, + })])?; + + let Some(batch) = subscriber.next_batch().await? else { + bail!("expected one batch from the polling log stream"); + }; + let records = batch.records(); + assert_eq!(records.len(), 1); + assert_eq!(records[0].context.source, InputSource::Poll); + assert_eq!(records[0].context.transaction_index, Some(7)); + assert_eq!(records[0].context.log_index, Some(7)); + assert!( + matches!(records[0].context.chain_status, ChainStatus::Included { ref block, confirmations: 0 } if block.number == 42 && block.hash == B256::repeat_byte(42)) + ); + assert!(matches!(&records[0].input, ReactiveInput::Log(actual) if actual == &log)); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +#[cfg(all(feature = "reactive-polling", not(feature = "reactive-ws")))] +async fn alloy_subscriber_auto_mode_uses_polling_when_ws_is_not_compiled() -> Result<()> { + let asserter = Asserter::new(); + let address = Address::repeat_byte(0xef); + let topic0 = keccak256(b"AutoMode(uint256)"); + let log = rpc_log(address, topic0, 50, 0); + + asserter.push_success(&U256::from(4)); + asserter.push_success(&vec![log.clone()]); + + let provider = ProviderBuilder::new().connect_mocked_client(asserter); + let mut subscriber = AlloySubscriber::new( + provider, + SubscriberMode::Auto, + SubscriberConfig { + hydrate_pending_transactions: false, + max_batch_size: 16, + ..SubscriberConfig::default() + }, + ); + subscriber.register_interests(&[ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(address).event_signature(topic0), + local_matcher: None, + route_key: None, + })])?; + + let Some(batch) = subscriber.next_batch().await? else { + bail!("expected auto mode to use polling and produce one batch"); + }; + + assert_eq!(batch.records().len(), 1); + assert!(matches!(&batch.records()[0].input, ReactiveInput::Log(actual) if actual == &log)); + assert_eq!(SubscriberMode::default(), SubscriberMode::Auto); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +#[cfg(feature = "reactive-polling")] +async fn alloy_subscriber_polling_pending_hashes_yield_pending_records() -> Result<()> { + let asserter = Asserter::new(); + let hash = B256::repeat_byte(0x55); + + asserter.push_success(&U256::from(2)); + asserter.push_success(&vec![hash]); + + let mut subscriber = polling_subscriber(asserter, 16); + subscriber.register_interests(&[ReactiveInterest::PendingTransactions( + PendingTxInterest::default(), + )])?; + + let Some(batch) = subscriber.next_batch().await? else { + bail!("expected one batch from the polling pending transaction stream"); + }; + let records = batch.records(); + assert_eq!(records.len(), 1); + assert_eq!(records[0].context.source, InputSource::Poll); + assert!(matches!( + records[0].context.chain_status, + ChainStatus::Pending + )); + assert!(matches!(records[0].input, ReactiveInput::PendingTxHash(actual) if actual == hash)); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +#[cfg(feature = "reactive-polling")] +async fn alloy_subscriber_removed_logs_yield_reorged_context() -> Result<()> { + let asserter = Asserter::new(); + let address = Address::repeat_byte(0x12); + let topic0 = keccak256(b"Removed(uint256)"); + let log = removed_rpc_log(address, topic0, 75, 2); + + asserter.push_success(&U256::from(5)); + asserter.push_success(&vec![log.clone()]); + + let mut subscriber = polling_subscriber(asserter, 16); + subscriber.register_interests(&[ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(address).event_signature(topic0), + local_matcher: None, + route_key: None, + })])?; + + let Some(batch) = subscriber.next_batch().await? else { + bail!("expected removed log batch"); + }; + let records = batch.records(); + assert_eq!(records.len(), 1); + assert!( + matches!(records[0].context.chain_status, ChainStatus::Reorged { ref dropped_from } if dropped_from.number == 75 && dropped_from.hash == B256::repeat_byte(75)) + ); + assert!(matches!(&records[0].input, ReactiveInput::Log(actual) if actual == &log)); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +#[cfg(feature = "reactive-polling")] +async fn alloy_subscriber_respects_max_batch_size_for_polled_logs() -> Result<()> { + let asserter = Asserter::new(); + let address = Address::repeat_byte(0xcd); + let topic0 = keccak256(b"Chunked(uint256)"); + let first = rpc_log(address, topic0, 100, 0); + let second = rpc_log(address, topic0, 100, 1); + + asserter.push_success(&U256::from(3)); + asserter.push_success(&vec![first.clone(), second.clone()]); + + let mut subscriber = polling_subscriber(asserter, 1); + subscriber.register_interests(&[ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(address).event_signature(topic0), + local_matcher: None, + route_key: None, + })])?; + + let Some(first_batch) = subscriber.next_batch().await? else { + bail!("expected first chunk"); + }; + let Some(second_batch) = subscriber.next_batch().await? else { + bail!("expected second chunk"); + }; + + assert_eq!(first_batch.records().len(), 1); + assert_eq!(second_batch.records().len(), 1); + assert!( + matches!(&first_batch.records()[0].input, ReactiveInput::Log(actual) if actual == &first) + ); + assert!( + matches!(&second_batch.records()[0].input, ReactiveInput::Log(actual) if actual == &second) + ); + + Ok(()) +} + +#[test] +#[cfg(feature = "reactive-polling")] +fn alloy_subscriber_polling_block_streams_are_explicitly_unsupported() -> Result<()> { + let mut polling = mock_subscriber(SubscriberMode::Polling); + let block_result = + polling.register_interests(&[ReactiveInterest::Blocks(BlockInterest::default())]); + assert!(matches!(block_result, Err(SubscriberError::Unsupported(_)))); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +#[cfg(feature = "reactive-polling")] +async fn alloy_subscriber_provider_errors_are_reported() -> Result<()> { + let asserter = Asserter::new(); + let address = Address::repeat_byte(0x34); + let topic0 = keccak256(b"ProviderError(uint256)"); + + let mut subscriber = polling_subscriber(asserter, 16); + subscriber.register_interests(&[ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(address).event_signature(topic0), + local_matcher: None, + route_key: None, + })])?; + + let result = subscriber.next_batch().await; + assert!(matches!(result, Err(SubscriberError::Provider(_)))); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +#[cfg(feature = "reactive-polling")] +async fn alloy_subscriber_reports_dropped_polling_filters() -> Result<()> { + let asserter = Asserter::new(); + let address = Address::repeat_byte(0x56); + let topic0 = keccak256(b"DroppedFilter(uint256)"); + + asserter.push_success(&U256::from(6)); + asserter.push_failure_msg("filter not found"); + + let mut subscriber = polling_subscriber(asserter, 16); + subscriber.register_interests(&[ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(address).event_signature(topic0), + local_matcher: None, + route_key: None, + })])?; + + let result = subscriber.next_batch().await; + assert!( + matches!(result, Err(SubscriberError::Provider(ref message)) if message.contains("terminated")), + "dropped provider filters must be surfaced instead of returning Ok(None): {result:?}" + ); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +async fn alloy_subscriber_zero_max_batch_size_is_rejected() -> Result<()> { + let provider = ProviderBuilder::new().connect_mocked_client(Asserter::new()); + let mut subscriber = AlloySubscriber::new( + provider, + SubscriberMode::Auto, + SubscriberConfig { + hydrate_pending_transactions: false, + max_batch_size: 0, + ..SubscriberConfig::default() + }, + ); + + let result = subscriber.register_interests(&[ReactiveInterest::PendingTransactions( + PendingTxInterest::default(), + )]); + + assert!( + result.is_err(), + "zero max_batch_size must be rejected instead of creating a subscriber that cannot make progress" + ); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +async fn alloy_subscriber_rejects_invalid_reconnect_config() -> Result<()> { + let defaults = SubscriberReconnectConfig::default(); + assert_eq!(defaults.initial_delay, Duration::ZERO); + assert_eq!(defaults.retry_delay, Duration::from_millis(250)); + assert_eq!(defaults.max_attempts, Some(3)); + + let provider = ProviderBuilder::new().connect_mocked_client(Asserter::new()); + let mut subscriber = AlloySubscriber::new( + provider, + SubscriberMode::Auto, + SubscriberConfig { + reconnect: SubscriberReconnectConfig { + retry_delay: Duration::from_secs(2), + max_delay: Duration::from_secs(1), + ..SubscriberReconnectConfig::default() + }, + ..SubscriberConfig::default() + }, + ); + + let result = subscriber.register_interests(&[ReactiveInterest::PendingTransactions( + PendingTxInterest::default(), + )]); + assert!(matches!(result, Err(SubscriberError::InvalidConfig(_)))); + + let provider = ProviderBuilder::new().connect_mocked_client(Asserter::new()); + let mut subscriber = AlloySubscriber::new( + provider, + SubscriberMode::Auto, + SubscriberConfig { + reconnect: SubscriberReconnectConfig { + max_attempts: Some(0), + ..SubscriberReconnectConfig::default() + }, + ..SubscriberConfig::default() + }, + ); + + let result = subscriber.register_interests(&[ReactiveInterest::PendingTransactions( + PendingTxInterest::default(), + )]); + assert!(matches!(result, Err(SubscriberError::InvalidConfig(_)))); + + Ok(()) +}