diff --git a/CHANGELOG.md b/CHANGELOG.md index 1310bc0..330b1f3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -123,6 +123,31 @@ pre-release development phases (see [`docs/ROADMAP.md`](docs/ROADMAP.md)). chain truth (correct **and** alarm) via the new `EvmCache::reconcile_slots`. A thin async `drive`/`LogSource` convenience layers the synchronous core over a stream. Generic core. +- **Reactive runtime** (`reactive` module, default-enabled) — a provider-neutral + handler pipeline for logs, block notifications, and pending transaction + signals. `ReactiveHandler`s are pure synchronous functions over + `ReactiveInput` + `ReactiveContext` + `StateView`; they emit `StateUpdate`s, + invalidations, resync requests, speculative requests, and hook signals. The + runtime deduplicates inputs by `InputRef`, orders canonical logs by + `(block_number, transaction_index, log_index)`, routes by `ReactiveInterest` + with Alloy `Filter`s and local matchers, validates pending inputs so they + cannot mutate canonical cache state, detects conflicting absolute writes for a + single input, applies canonical mutations through `EvmCache::apply_updates`, + and dispatches `ReactiveReport`s to hooks after mutation phases. + `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. +- **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 + `StorageBatchFetchFn`, applies successful values as `StateUpdate::slot` + updates, and reports requested targets, applied updates, the resulting + `StateDiff`, and per-target failures in `ResyncReport`. `ResyncFailureKind` + gives downstream retry policy and metrics a stable failure classification. + Account-field resyncs remain explicitly unsupported until a provider-neutral + account fetch callback exists. Generic core. - **`StateUpdate::SlotMasked`** (`state_update`, Phase 4) — a cold-aware read-modify-write *masked* slot write (`new = (old & !mask) | (value & mask)`) with the `StateUpdate::slot_masked` constructor, so a pure decoder can update diff --git a/Cargo.toml b/Cargo.toml index 845b370..307864b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,6 +15,10 @@ documentation = "https://docs.rs/evm-fork-cache" # ancestor-directory workspace and gives it its own Cargo.lock. [workspace] +[features] +default = ["reactive"] +reactive = [] + [dependencies] alloy-consensus = "1.1.2" alloy-contract = "1.0.38" diff --git a/README.md b/README.md index 0c6d13b..bc76e7c 100644 --- a/README.md +++ b/README.md @@ -38,10 +38,10 @@ around three capabilities that target exactly this workload: > **Maturity.** This crate is **pre-1.0** and under active development against a > [phased roadmap](docs/ROADMAP.md). Capabilities (1) and (3) above are -> implemented today. Capability (2) has the targeted writer primitives and the -> event-to-state reader pipeline; a production WebSocket transport remains -> consumer-provided. The public API still changes between minor versions — see -> [Stability](#stability). +> implemented today. Capability (2) has the targeted writer primitives, the +> event-to-state reader pipeline, and a default-enabled reactive handler runtime; +> live network subscription driving remains consumer-provided. The public API +> still changes between minor versions — see [Stability](#stability). ## What it provides today @@ -60,6 +60,18 @@ around three capabilities that target exactly this workload: 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. +- **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 + inputs, deduplicates and orders canonical logs, validates pending semantics, + applies canonical cache mutations through `EvmCache::apply_updates`, and + can optionally execute storage resync requests through the cache's + 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. - **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 diff --git a/src/lib.rs b/src/lib.rs index 19f65ae..0208e04 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -54,6 +54,10 @@ //! `StateView` / `DecoderRegistry` decode an on-chain `Log` into `StateUpdate`s, //! and `EventPipeline` ingests, reorg-purges, and reconciles a block's logs. //! Ships an ERC-20 `Transfer` decoder plus traits for external decoders. +//! - `reactive` — default-enabled provider-neutral handler runtime for logs, +//! blocks, and pending transaction signals. Pure handlers emit `StateUpdate`s, +//! invalidations, resync requests, speculative signals, and hooks; the runtime +//! validates and applies canonical cache mutations. //! - [`inspector`] — an [`Inspector`](revm::Inspector) that captures ERC20 //! `Transfer` events to reconstruct balance deltas from a simulation. //! - [`multicall`] — batched read-only calls through Multicall3. @@ -110,6 +114,8 @@ pub mod freshness; pub mod inspector; pub mod multicall; pub mod prefetch_registry; +#[cfg(feature = "reactive")] +pub mod reactive; pub mod state_update; pub use access_set::StorageAccessList; diff --git a/src/reactive/mod.rs b/src/reactive/mod.rs new file mode 100644 index 0000000..7ad9ddd --- /dev/null +++ b/src/reactive/mod.rs @@ -0,0 +1,2141 @@ +//! Protocol-neutral reactive runtime for cache state effects. +//! +//! The reactive runtime generalizes the log-only [`events`](crate::events) +//! pipeline into a handler pipeline that can ingest logs, block notifications, +//! and pending transaction signals. Handlers remain pure synchronous functions: +//! they read through [`StateView`], return structured +//! [`ReactiveEffect`] values, and let the runtime validate and commit cache +//! mutations through [`StateUpdate`]. +//! +//! This module intentionally contains no protocol, AMM, strategy, signing, or +//! transaction-submission concepts. Downstream crates can layer those domains on +//! top by implementing [`ReactiveHandler`] and [`ReactiveHook`]. + +use std::{ + any::Any, + borrow::Cow, + collections::{HashMap, HashSet}, + fmt, + future::Future, + hash::Hash, + marker::PhantomData, + pin::Pin, + sync::Arc, +}; + +use alloy_consensus::{BlockHeader as _, Transaction as _}; +use alloy_eips::BlockId; +use alloy_network::{ + Ethereum, Network, + primitives::{ + BlockResponse as _, HeaderResponse as _, TransactionResponse as TransactionResponseTrait, + }, +}; +use alloy_primitives::{Address, B256, Bytes, U256}; +use alloy_rpc_types_eth::{Filter, FilterSet, Log}; + +use crate::{ + cache::EvmCache, + events::{EventDecoder, StateView}, + state_update::{AccountPatch, PurgeScope, StateDiff, StateUpdate}, +}; + +/// Input accepted by the reactive runtime. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum ReactiveInput { + /// A canonical or removed EVM log, using Alloy's RPC log type. + Log(Log), + /// A block header response for header-oriented handlers. + BlockHeader(N::HeaderResponse), + /// A full block response for block handlers that need transaction bodies. + FullBlock(N::BlockResponse), + /// A pending transaction hash. + PendingTxHash(B256), + /// A full pending transaction body. + PendingTx(N::TransactionResponse), +} + +/// Context supplied with each [`ReactiveInput`]. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ReactiveContext { + /// Chain id, when known. + pub chain_id: Option, + /// Where the input came from. + pub source: InputSource, + /// Lifecycle status of the input. + pub chain_status: ChainStatus, + /// Block metadata associated with the input, when known. + pub block: Option, + /// Transaction index for log or transaction inputs. + pub transaction_index: Option, + /// Log index for log inputs. + pub log_index: Option, +} + +/// Minimal block identity carried through reports. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct BlockRef { + /// Block number. + pub number: u64, + /// Block hash. + pub hash: B256, + /// Parent hash, when known. + pub parent_hash: Option, + /// Block timestamp, when known. + pub timestamp: Option, +} + +/// Lifecycle status for an input. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum ChainStatus { + /// The input is mempool-only and must not mutate canonical cache state. + Pending, + /// The input is included in a block with a confirmation count. + Included { + /// Included block. + block: BlockRef, + /// Confirmation count. + confirmations: u64, + }, + /// The input is in the chain's safe head. + Safe { + /// Safe block. + block: BlockRef, + }, + /// The input is in the finalized head. + Finalized { + /// Finalized block. + block: BlockRef, + }, + /// The input was dropped by a reorg. + Reorged { + /// Block the input was dropped from. + dropped_from: BlockRef, + }, +} + +/// Source of an input batch. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub enum InputSource { + /// Caller-supplied batch. + Batch, + /// Live subscription stream. + Subscription, + /// Polling subscriber. + Poll, + /// Historical backfill. + Backfill, + /// Test or synthetic input. + Synthetic, +} + +/// Stable identity used for input deduplication and reports. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub enum InputRef { + /// Stable log identity. + Log { + /// Chain id, when known. + chain_id: Option, + /// Block hash containing the log. + block_hash: B256, + /// Transaction hash that emitted the log. + transaction_hash: B256, + /// Log index within the block. + log_index: u64, + }, + /// Stable pending transaction identity. + PendingTx { + /// Chain id, when known. + chain_id: Option, + /// Transaction hash. + hash: B256, + }, + /// Stable block identity. + Block { + /// Chain id, when known. + chain_id: Option, + /// Block hash. + hash: B256, + /// Block number. + number: u64, + }, +} + +/// Reliability of state effects emitted by a handler. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub enum StateEffectQuality { + /// Effects are exact from the input alone. + ExactFromInput, + /// Effects were applied, but follow-up resync is pending. + AppliedWithPendingResync, + /// Effects came from authoritative resync. + ResyncedAuthoritatively, + /// State requires repair before it should be trusted. + RequiresRepair, + /// No canonical state effect was emitted. + NoStateEffect, +} + +/// Identifier for a reactive handler. +#[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)] +pub struct HandlerId(String); + +impl HandlerId { + /// Create a handler id. + pub fn new(id: impl Into) -> Self { + Self(id.into()) + } + + /// Return the id as a string slice. + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl fmt::Display for HandlerId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + self.0.fmt(f) + } +} + +/// Lightweight report label. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct ReportTag { + /// Label key. + pub key: String, + /// Label value. + pub value: String, +} + +impl ReportTag { + /// Create a report tag. + pub fn new(key: impl Into, value: impl Into) -> Self { + Self { + key: key.into(), + value: value.into(), + } + } +} + +/// Domain-neutral hook signal emitted by a handler. +#[derive(Clone)] +pub struct HookSignal { + /// Signal namespace owned by the caller. + pub namespace: Cow<'static, str>, + /// Signal kind within the namespace. + pub kind: Cow<'static, str>, + /// Additional labels for routing or observability. + pub labels: Vec, + /// Optional in-process typed payload. + pub payload: Option>, +} + +impl fmt::Debug for HookSignal { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("HookSignal") + .field("namespace", &self.namespace) + .field("kind", &self.kind) + .field("labels", &self.labels) + .field("payload", &self.payload.as_ref().map(|_| "")) + .finish() + } +} + +/// Effect emitted by a [`ReactiveHandler`]. +#[derive(Clone, Debug)] +pub enum ReactiveEffect { + /// Canonical cache mutation applied through [`EvmCache::apply_updates`]. + StateUpdate(StateUpdate), + /// Request for authoritative state repair. + Resync(ResyncRequest), + /// Rich invalidation request lowered to [`StateUpdate::Purge`]. + Invalidate(InvalidationRequest), + /// Hook signal dispatched after committed mutation phases. + Hook(HookSignal), + /// Speculative signal for mempool or downstream work. + Speculative(SpeculativeRequest), +} + +/// Handler output for a single input. +#[derive(Clone, Debug)] +pub struct HandlerOutcome { + /// Effects emitted by the handler. + pub effects: Vec, + /// Reliability of emitted state effects. + pub quality: StateEffectQuality, + /// Labels copied into reports. + pub tags: Vec, +} + +impl HandlerOutcome { + /// Construct an empty outcome with the supplied quality. + pub fn empty(quality: StateEffectQuality) -> Self { + Self { + effects: Vec::new(), + quality, + tags: Vec::new(), + } + } +} + +/// One input and its execution context. +#[derive(Clone, Debug)] +pub struct ReactiveInputRecord { + /// Input value. + pub input: ReactiveInput, + /// Input context. + pub context: ReactiveContext, +} + +impl ReactiveInputRecord { + /// Create an input record. + pub fn new(input: ReactiveInput, context: ReactiveContext) -> Self { + Self { input, context } + } + + /// Compute the stable input reference used for deduplication. + pub fn input_ref(&self) -> InputRef { + input_ref(&self.input, &self.context) + } +} + +/// Batch of reactive input records. +#[derive(Clone, Debug)] +pub struct ReactiveInputBatch { + records: Vec>, +} + +impl ReactiveInputBatch { + /// Create a batch from records. + pub fn new(records: Vec>) -> Self { + Self { records } + } + + /// Borrow the records in this batch. + pub fn records(&self) -> &[ReactiveInputRecord] { + &self.records + } + + /// Consume the batch into its records. + pub fn into_records(self) -> Vec> { + self.records + } +} + +/// Pure synchronous handler for reactive inputs. +pub trait ReactiveHandler: Send + Sync { + /// Stable handler id. + fn id(&self) -> HandlerId; + + /// Interests used by subscribers and the local router. + fn interests(&self) -> Vec>; + + /// Handle one input against a read-only cache view. + fn handle( + &self, + ctx: &ReactiveContext, + input: &ReactiveInput, + state: &dyn StateView, + ) -> Result; +} + +/// Hook invoked after reports are built and cache mutation phases have ended. +pub trait ReactiveHook: Send + Sync { + /// Observe a runtime report. + fn on_report(&self, report: Arc>); +} + +/// Reactive subscription interest. +#[allow(clippy::large_enum_variant)] +#[derive(Clone)] +pub enum ReactiveInterest { + /// Log interest. + Logs(LogInterest), + /// Block interest. + Blocks(BlockInterest), + /// Pending transaction interest. + PendingTransactions(PendingTxInterest), +} + +impl fmt::Debug for ReactiveInterest { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Logs(interest) => f.debug_tuple("Logs").field(interest).finish(), + Self::Blocks(interest) => f.debug_tuple("Blocks").field(interest).finish(), + Self::PendingTransactions(interest) => f + .debug_tuple("PendingTransactions") + .field(interest) + .finish(), + } + } +} + +/// Interest in logs. +#[derive(Clone)] +pub struct LogInterest { + /// Provider-side filter. + pub provider_filter: Filter, + /// Optional local matcher for predicates providers cannot express. + pub local_matcher: Option>, + /// Optional route-key extraction strategy. + pub route_key: Option, +} + +impl LogInterest { + /// Return true if the log matches both the provider filter and local matcher. + pub fn matches(&self, log: &Log) -> bool { + self.provider_filter.rpc_matches(log) + && self + .local_matcher + .as_ref() + .is_none_or(|matcher| matcher.matches(log)) + } + + /// Extract the route key for a matching log, if configured. + pub fn route_key(&self, log: &Log) -> Option { + self.route_key.as_ref().and_then(|spec| spec.extract(log)) + } +} + +impl fmt::Debug for LogInterest { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("LogInterest") + .field("provider_filter", &self.provider_filter) + .field( + "local_matcher", + &self.local_matcher.as_ref().map(|_| ""), + ) + .field("route_key", &self.route_key) + .finish() + } +} + +/// Local log predicate. +pub trait LogMatcher: Send + Sync { + /// Return true when the log should be routed to the handler. + fn matches(&self, log: &Log) -> bool; +} + +/// Route-key extraction strategy for logs. +#[derive(Clone)] +pub enum RouteKeySpec { + /// Route by emitting address. + EmitterAddress, + /// Route by indexed topic. + Topic { + /// Topic index. + index: usize, + }, + /// Route by a byte slice in log data. + DataSlice { + /// Byte offset in the data payload. + offset: usize, + /// Number of bytes to copy. + len: usize, + }, + /// Custom extractor. + Custom(Arc), +} + +impl RouteKeySpec { + /// Extract a route key from a log. + pub fn extract(&self, log: &Log) -> Option { + match self { + Self::EmitterAddress => Some(RouteKey::Address(log.address())), + Self::Topic { index } => log.topics().get(*index).copied().map(RouteKey::Bytes32), + Self::DataSlice { offset, len } => { + let data = log.inner.data.data.as_ref(); + let end = offset.checked_add(*len)?; + data.get(*offset..end) + .map(|bytes| RouteKey::Bytes(bytes.to_vec())) + } + Self::Custom(extractor) => extractor.extract(log), + } + } +} + +impl fmt::Debug for RouteKeySpec { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::EmitterAddress => f.write_str("EmitterAddress"), + Self::Topic { index } => f.debug_struct("Topic").field("index", index).finish(), + Self::DataSlice { offset, len } => f + .debug_struct("DataSlice") + .field("offset", offset) + .field("len", len) + .finish(), + Self::Custom(_) => f.write_str("Custom()"), + } + } +} + +/// Extracts custom route keys from logs. +pub trait RouteKeyExtractor: Send + Sync { + /// Extract a route key. + fn extract(&self, log: &Log) -> Option; +} + +/// Extracted route key. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub enum RouteKey { + /// Address key. + Address(Address), + /// 32-byte key. + Bytes32(B256), + /// Arbitrary bytes key. + Bytes(Vec), +} + +/// Exact log route selected by [`ReactiveRegistry::route_log`]. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ReactiveLogRoute { + /// Handler whose log interest matched. + pub handler_id: HandlerId, + /// Optional route key extracted from the matching log interest. + pub route_key: Option, +} + +/// Interest in block inputs. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct BlockInterest { + /// Block input mode. + pub mode: BlockInterestMode, +} + +impl Default for BlockInterest { + fn default() -> Self { + Self { + mode: BlockInterestMode::Header, + } + } +} + +/// Block subscription mode. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub enum BlockInterestMode { + /// Header-only block input. + Header, + /// Full block input. + FullBlock, +} + +/// Interest in pending transaction inputs. +#[derive(Clone)] +pub struct PendingTxInterest { + /// Whether the handler requires full transaction bodies. + pub full_transactions: bool, + /// Sender matcher. + pub from: AddressMatcher, + /// Recipient matcher. + pub to: AddressMatcher, + /// Calldata selector matcher. + pub selectors: SelectorMatcher, + /// Optional local transaction matcher. + pub local_matcher: Option>>, +} + +impl Default for PendingTxInterest { + fn default() -> Self { + Self { + full_transactions: false, + from: AddressMatcher::Any, + to: AddressMatcher::Any, + selectors: SelectorMatcher::Any, + local_matcher: None, + } + } +} + +impl PendingTxInterest { + fn matches_hash_only(&self) -> bool { + !self.full_transactions + && self.from.is_any() + && self.to.is_any() + && self.selectors.is_any() + && self.local_matcher.is_none() + } + + fn matches_tx(&self, tx: &N::TransactionResponse) -> bool { + self.from.matches(tx.from()) + && self.to.matches_option(tx.to()) + && self.selectors.matches(tx.input()) + && self + .local_matcher + .as_ref() + .is_none_or(|matcher| matcher.matches(tx)) + } +} + +impl fmt::Debug for PendingTxInterest { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("PendingTxInterest") + .field("full_transactions", &self.full_transactions) + .field("from", &self.from) + .field("to", &self.to) + .field("selectors", &self.selectors) + .field( + "local_matcher", + &self.local_matcher.as_ref().map(|_| ""), + ) + .finish() + } +} + +/// Address matching helper for pending transaction interests. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub enum AddressMatcher { + /// Match every address. + Any, + /// Match one address. + Exact(Address), + /// Match any address in the list. + AnyOf(Vec
), +} + +impl AddressMatcher { + /// Return true when the matcher is unconstrained. + pub fn is_any(&self) -> bool { + matches!(self, Self::Any) + } + + /// Match a present address. + pub fn matches(&self, address: Address) -> bool { + match self { + Self::Any => true, + Self::Exact(expected) => *expected == address, + Self::AnyOf(addresses) => addresses.contains(&address), + } + } + + /// Match an optional address. + pub fn matches_option(&self, address: Option
) -> bool { + match (self, address) { + (Self::Any, _) => true, + (_, Some(address)) => self.matches(address), + _ => false, + } + } +} + +/// Calldata selector matching helper. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub enum SelectorMatcher { + /// Match every selector. + Any, + /// Match any selector in the list. + AnyOf(Vec<[u8; 4]>), +} + +impl SelectorMatcher { + /// Return true when the matcher is unconstrained. + pub fn is_any(&self) -> bool { + matches!(self, Self::Any) + } + + /// Match calldata bytes. + pub fn matches(&self, input: &Bytes) -> bool { + match self { + Self::Any => true, + Self::AnyOf(selectors) => input + .get(..4) + .and_then(|bytes| bytes.try_into().ok()) + .is_some_and(|selector| selectors.contains(&selector)), + } + } +} + +/// Local predicate over a full pending transaction. +pub trait PendingTxMatcher: Send + Sync { + /// Return true when the transaction should be routed to the handler. + fn matches(&self, tx: &N::TransactionResponse) -> bool; +} + +/// Request for authoritative state repair. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ResyncRequest { + /// Resync id. + pub id: ResyncId, + /// Reason for the request. + pub reason: ResyncReason, + /// Block selection for the read. + pub block: ResyncBlock, + /// Targets to resync. + pub targets: Vec, + /// Scheduling priority. + pub priority: ResyncPriority, +} + +/// Resync id. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct ResyncId(String); + +impl ResyncId { + /// Create a resync id. + pub fn new(id: impl Into) -> Self { + Self(id.into()) + } +} + +/// Reason for a resync request. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub enum ResyncReason { + /// Handler requested repair. + HandlerRequested, + /// State effect could not be applied completely. + SkippedStateEffect, + /// Caller-defined reason. + Custom(String), +} + +/// Block target for a resync. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub enum ResyncBlock { + /// Latest block. + Latest, + /// Safe head. + Safe, + /// Finalized head. + Finalized, + /// Block number. + Number(u64), + /// Block hash and number. + Hash { + /// Block number. + number: u64, + /// Block hash. + hash: B256, + /// Require the hash to still be canonical. + require_canonical: bool, + }, +} + +/// State target for a resync. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub enum ResyncTarget { + /// One storage slot. + StorageSlot { + /// Contract address. + address: Address, + /// Storage slot. + slot: U256, + }, + /// Multiple storage slots on one contract. + StorageSlots { + /// Contract address. + address: Address, + /// Storage slots. + slots: Vec, + }, + /// Account fields. + Account { + /// Account address. + address: Address, + /// Fields to resync. + fields: AccountFieldMask, + }, +} + +/// Account fields requested by a resync. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)] +pub struct AccountFieldMask { + /// Balance field. + pub balance: bool, + /// Nonce field. + pub nonce: bool, + /// Code field. + pub code: bool, +} + +/// Resync priority. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash, PartialOrd, Ord)] +pub enum ResyncPriority { + /// Low priority. + Low, + /// Normal priority. + #[default] + Normal, + /// High priority. + High, +} + +/// Rich invalidation request lowered to [`StateUpdate::Purge`]. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct InvalidationRequest { + /// Purge scope. + pub scope: PurgeScope, + /// Address to purge. + pub address: Address, + /// Reason for reporting. + pub reason: InvalidationReason, +} + +/// Invalidation reason. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub enum InvalidationReason { + /// Handler requested invalidation. + HandlerRequested, + /// Reorg invalidation. + Reorg, + /// Caller-defined reason. + Custom(String), +} + +/// Speculative signal emitted by handlers. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SpeculativeRequest { + /// Speculative request id. + pub id: SpeculativeId, + /// Input that triggered the request. + pub input_ref: InputRef, + /// Labels for downstream routing. + pub labels: Vec, +} + +/// Speculative request id. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct SpeculativeId(String); + +impl SpeculativeId { + /// Create a speculative id. + pub fn new(id: impl Into) -> Self { + Self(id.into()) + } +} + +/// Configuration for [`ReactiveRuntime`]. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ReactiveConfig { + /// Hook backpressure policy for future async dispatchers. + pub hook_backpressure: HookBackpressure, + /// Reorg journal depth reserved for future rollback support. + pub journal_depth: usize, +} + +impl Default for ReactiveConfig { + fn default() -> Self { + Self { + hook_backpressure: HookBackpressure::Block, + journal_depth: 64, + } + } +} + +/// Hook backpressure policy. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub enum HookBackpressure { + /// Block the producer until hooks are accepted. + Block, + /// Drop the newest report under pressure. + DropNewest, + /// Drop the oldest report under pressure. + DropOldest, + /// Return an error under pressure. + Error, +} + +/// Runtime report. +#[derive(Clone, Debug)] +pub enum ReactiveReport { + /// Input was accepted after deduplication. + Input(InputReport), + /// Handlers produced outcomes. + Decoded(DecodedReport), + /// Direct state effects were applied. + Applied(AppliedReport), + /// Resync request was scheduled or completed. + Resynced(ResyncReport), + /// Block-level processing completed. + BlockCommitted(BlockReport), + /// Reorg processing report. + Reorg(ReorgReport), + /// Runtime or handler error. + Error(ReactiveErrorReport), +} + +/// Input acceptance report. +#[derive(Clone, Debug)] +pub struct InputReport { + /// Input reference. + pub input_ref: InputRef, + /// Input context. + pub context: ReactiveContext, + /// Network marker. + pub _network: PhantomData, +} + +/// Decoding report. +#[derive(Clone, Debug)] +pub struct DecodedReport { + /// Input reference. + pub input_ref: InputRef, + /// Handler ids that matched the input. + pub handler_ids: Vec, + /// Network marker. + pub _network: PhantomData, +} + +/// Applied state report. +#[derive(Clone, Debug)] +pub struct AppliedReport { + /// Input reference. + pub input_ref: InputRef, + /// Handler that produced the applied effects. + pub handler_id: HandlerId, + /// State effect quality. + pub quality: StateEffectQuality, + /// Labels emitted by the handler. + pub tags: Vec, + /// Merged state diff from applied updates and invalidations. + pub diff: StateDiff, + /// State updates applied through the cache. + pub state_updates: Vec, + /// Invalidation requests lowered to purge updates. + pub invalidations: Vec, + /// Resync requests surfaced for a scheduler. + pub resyncs: Vec, + /// Speculative requests surfaced for downstream users. + pub speculative: Vec, + /// Hook signals emitted by the handler. + pub hook_signals: Vec, + /// Network marker. + pub _network: PhantomData, +} + +/// Resync reporting scaffold. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct ResyncReport { + /// Requests considered by the resync execution pass. + pub requested: Vec, + /// Authoritative state updates built from successful resync fetches. + pub state_updates: Vec, + /// Diff returned by applying [`state_updates`](Self::state_updates). + pub diff: StateDiff, + /// Targets that could not be resynced. + pub failed: Vec, +} + +/// One resync target that could not be fetched or applied. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ResyncFailure { + /// Request that produced the failed target. + pub request_id: ResyncId, + /// Block selection used for the failed target. + pub block: ResyncBlock, + /// Target that could not be resynced. + pub target: ResyncTarget, + /// Stable failure classification for retry policy and metrics. + pub kind: ResyncFailureKind, + /// Human-readable failure reason. + pub message: String, +} + +/// Stable classification for a failed resync target. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub enum ResyncFailureKind { + /// A storage target could not be fetched because no storage batch fetcher is configured. + MissingStorageFetcher, + /// The storage batch fetcher returned an error for the requested slot. + StorageFetchFailed, + /// The storage batch fetcher did not return a result for the requested slot. + StorageFetchOmitted, + /// Account-field resync is not supported by the current provider-neutral cache seam. + UnsupportedAccountTarget, +} + +/// Block processing report. +#[derive(Clone, Debug)] +pub struct BlockReport { + /// Block reference, when known. + pub block: Option, + /// Input references committed for the block. + pub inputs: Vec, + /// Network marker. + pub _network: PhantomData, +} + +/// Reorg reporting scaffold. +#[derive(Clone, Debug)] +pub struct ReorgReport { + /// Dropped block, when known. + pub dropped: Option, + /// Network marker. + pub _network: PhantomData, +} + +/// Error report scaffold. +#[derive(Clone, Debug)] +pub struct ReactiveErrorReport { + /// Input associated with the error, when known. + pub input_ref: Option, + /// Error message. + pub message: String, + /// Network marker. + pub _network: PhantomData, +} + +/// Batch report returned by [`ReactiveRuntime::ingest_batch`] and +/// [`ReactiveRuntime::ingest_batch_with_resync`]. +#[derive(Clone, Debug)] +pub struct ReactiveBatchReport { + /// Applied reports in commit order. + pub applied: Vec>, + /// Resync requests surfaced during the batch. + pub resyncs: Vec, + /// Speculative requests surfaced during the batch. + pub speculative: Vec, + /// Hook reports dispatched after mutation phases. + pub reports: Vec>>, +} + +impl Default for ReactiveBatchReport { + fn default() -> Self { + Self { + applied: Vec::new(), + resyncs: Vec::new(), + speculative: Vec::new(), + reports: Vec::new(), + } + } +} + +/// Error returned by a handler. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct HandlerError { + message: String, +} + +impl HandlerError { + /// Create a handler error from a message. + pub fn new(message: impl Into) -> Self { + Self { + message: message.into(), + } + } +} + +impl fmt::Display for HandlerError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + self.message.fmt(f) + } +} + +impl std::error::Error for HandlerError {} + +impl From for HandlerError { + fn from(message: String) -> Self { + Self::new(message) + } +} + +impl From<&str> for HandlerError { + fn from(message: &str) -> Self { + Self::new(message) + } +} + +/// Runtime error. +#[derive(Debug, thiserror::Error)] +pub enum ReactiveError { + /// Handler returned an error. + #[error("handler `{handler_id}` failed: {source}")] + HandlerFailed { + /// Handler id. + handler_id: HandlerId, + /// Handler error. + source: HandlerError, + }, + /// Multiple handlers emitted incompatible absolute writes for one input. + #[error( + "conflicting effects for input {input_ref:?} on target {target:?}: `{first}` vs `{second}`" + )] + ConflictingEffects { + /// Input reference. + input_ref: Box, + /// Conflicting target. + target: Box, + /// First handler id. + first: HandlerId, + /// Second handler id. + second: HandlerId, + }, + /// Pending inputs attempted to mutate canonical cache state. + #[error( + "pending input {input_ref:?} emitted invalid canonical effect `{effect_kind}` from `{handler_id}`" + )] + InvalidPendingEffect { + /// Input reference. + input_ref: Box, + /// Handler id. + handler_id: HandlerId, + /// Effect kind. + effect_kind: &'static str, + }, + /// Registration error. + #[error(transparent)] + Register(#[from] RegisterError), +} + +/// Handler registration error. +#[derive(Debug, thiserror::Error)] +pub enum RegisterError { + /// Duplicate handler id. + #[error("handler id `{0}` is already registered")] + DuplicateHandler(HandlerId), +} + +/// Absolute write target used for conflict reports. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub enum EffectTarget { + /// Storage slot target. + StorageSlot { + /// Contract address. + address: Address, + /// Storage slot. + slot: U256, + }, + /// Account balance target. + AccountBalance { + /// Account address. + address: Address, + }, + /// Account nonce target. + AccountNonce { + /// Account address. + address: Address, + }, + /// Account code target. + AccountCode { + /// Account address. + address: Address, + }, + /// Masked storage slot target. + MaskedStorageSlot { + /// Contract address. + address: Address, + /// Storage slot. + slot: U256, + /// Bit mask. + mask: U256, + }, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +enum AbsoluteValue { + U256(U256), + U64(u64), + Bytes(Bytes), +} + +/// Reactive runtime. +pub struct ReactiveRuntime { + registry: ReactiveRegistry, + hooks: Vec>>, + config: ReactiveConfig, +} + +/// Registry and router for provider-neutral reactive handlers. +/// +/// The registry stores pure [`ReactiveHandler`]s in registration order, exposes +/// consolidated provider-side log filters for subscription setup, and routes +/// provider logs back to the exact matching log interests. Consolidated filters +/// may be safe supersets; [`Self::route_log`] always re-checks the original +/// [`LogInterest`] and its local matcher before returning a route. +pub struct ReactiveRegistry { + handlers: Vec>, +} + +struct RegisteredHandler { + id: HandlerId, + handler: Arc>, + interests: Vec>, +} + +impl Default for ReactiveRegistry { + fn default() -> Self { + Self::new() + } +} + +impl ReactiveRegistry { + /// Create an empty registry. + pub fn new() -> Self { + Self { + handlers: Vec::new(), + } + } + + /// Register a handler, preserving registration order. + /// + /// Duplicate handler ids are rejected with + /// [`RegisterError::DuplicateHandler`]. + pub fn register_handler( + &mut self, + handler: Arc>, + ) -> Result<(), RegisterError> { + let id = handler.id(); + if self.handlers.iter().any(|registered| registered.id == id) { + return Err(RegisterError::DuplicateHandler(id)); + } + let interests = handler.interests(); + self.handlers.push(RegisteredHandler { + id, + handler, + interests, + }); + Ok(()) + } + + /// Return all registered interests in handler registration order. + pub fn interests(&self) -> Vec> { + self.handlers + .iter() + .flat_map(|handler| handler.interests.clone()) + .collect() + } + + /// Return consolidated provider-side log filters. + /// + /// Filters are emitted in deterministic first-registration order by + /// compatible block option. Within each returned filter, address and topic + /// sets are unioned independently, which can intentionally overfetch. Use + /// [`Self::route_log`] to enforce the exact original [`LogInterest`]s. + pub fn log_subscription_filters(&self) -> Vec { + let mut filters = Vec::new(); + for interest in self.log_interests() { + merge_log_subscription_filter(&mut filters, &interest.provider_filter); + } + filters + } + + /// Route a log to exact matching handler interests. + /// + /// Routes are returned in handler registration order. Each handler appears + /// at most once for a log, using the first matching log interest declared by + /// that handler. + pub fn route_log(&self, log: &Log) -> Vec { + self.handlers + .iter() + .filter_map(|handler| handler.route_log(log)) + .collect() + } + + fn handlers(&self) -> &[RegisteredHandler] { + &self.handlers + } + + fn log_interests(&self) -> impl Iterator { + self.handlers.iter().flat_map(|handler| { + handler + .interests + .iter() + .filter_map(|interest| match interest { + ReactiveInterest::Logs(interest) => Some(interest), + ReactiveInterest::Blocks(_) | ReactiveInterest::PendingTransactions(_) => None, + }) + }) + } +} + +impl ReactiveRuntime { + /// Create an empty runtime. + pub fn new(config: ReactiveConfig) -> Self { + Self { + registry: ReactiveRegistry::new(), + hooks: Vec::new(), + config, + } + } + + /// Register a handler. + pub fn register_handler( + &mut self, + handler: Arc>, + ) -> Result<(), RegisterError> { + self.registry.register_handler(handler) + } + + /// Register a hook. + pub fn register_hook(&mut self, hook: Arc>) -> Result<(), RegisterError> { + self.hooks.push(hook); + Ok(()) + } + + /// Return all registered interests in handler registration order. + pub fn interests(&self) -> Vec> { + self.registry.interests() + } + + /// Ingest a batch, apply valid direct state effects, and dispatch reports. + pub fn ingest_batch( + &mut self, + cache: &mut EvmCache, + batch: ReactiveInputBatch, + ) -> Result, ReactiveError> { + let batch_report = self.ingest_batch_direct(cache, batch)?; + self.dispatch_reports(&batch_report.reports); + let _ = &self.config; + Ok(batch_report) + } + + /// Ingest a batch, then execute surfaced storage resync requests. + /// + /// This entrypoint preserves [`ingest_batch`](Self::ingest_batch) behavior for + /// direct handler effects, then runs a synchronous resync phase over the + /// collected [`ResyncRequest`]s. Storage targets are fetched through + /// [`EvmCache::storage_batch_fetcher`] grouped by [`ResyncBlock`], successful + /// values are applied as [`StateUpdate::slot`] updates through + /// [`EvmCache::apply_updates`], and unsupported or failed targets are reported + /// in [`ResyncReport::failed`]. It does not start subscribers, background + /// workers, reorg journals, or network transport. + pub fn ingest_batch_with_resync( + &mut self, + cache: &mut EvmCache, + batch: ReactiveInputBatch, + ) -> Result, ReactiveError> { + let mut batch_report = self.ingest_batch_direct(cache, batch)?; + + if !batch_report.resyncs.is_empty() { + let resync_report = execute_resync_requests(cache, &batch_report.resyncs); + batch_report + .reports + .push(Arc::new(ReactiveReport::Resynced(resync_report))); + } + + self.dispatch_reports(&batch_report.reports); + let _ = &self.config; + Ok(batch_report) + } + + fn ingest_batch_direct( + &self, + cache: &mut EvmCache, + batch: ReactiveInputBatch, + ) -> Result, ReactiveError> { + let records = sort_records(dedupe_records(batch.into_records())); + + let mut batch_report = ReactiveBatchReport::default(); + let mut reports_to_dispatch = Vec::new(); + + for record in records { + let input_ref = record.input_ref(); + reports_to_dispatch.push(Arc::new(ReactiveReport::Input(InputReport { + input_ref, + context: record.context.clone(), + _network: PhantomData, + }))); + + let executions = self.execute_handlers(cache, &record, input_ref)?; + if executions.is_empty() { + continue; + } + + reports_to_dispatch.push(Arc::new(ReactiveReport::Decoded(DecodedReport { + input_ref, + handler_ids: executions + .iter() + .map(|execution| execution.handler_id.clone()) + .collect(), + _network: PhantomData, + }))); + + detect_conflicts(input_ref, &executions)?; + + for execution in executions { + let diff = if execution.state_updates.is_empty() { + StateDiff::default() + } else { + cache.apply_updates(&execution.state_updates) + }; + + batch_report + .resyncs + .extend(execution.resyncs.iter().cloned()); + batch_report + .speculative + .extend(execution.speculative.iter().cloned()); + + let applied = AppliedReport { + input_ref, + handler_id: execution.handler_id, + quality: execution.quality, + tags: execution.tags, + diff, + state_updates: execution.state_updates, + invalidations: execution.invalidations, + resyncs: execution.resyncs, + speculative: execution.speculative, + hook_signals: execution.hook_signals, + _network: PhantomData, + }; + let report = Arc::new(ReactiveReport::Applied(applied.clone())); + reports_to_dispatch.push(report); + batch_report.applied.push(applied); + } + } + + batch_report.reports = reports_to_dispatch; + Ok(batch_report) + } + + fn execute_handlers( + &self, + cache: &EvmCache, + record: &ReactiveInputRecord, + input_ref: InputRef, + ) -> Result, ReactiveError> { + let mut executions = Vec::new(); + for registered in self.registry.handlers() { + if !registered.matches(&record.input) { + continue; + } + + let outcome = registered + .handler + .handle(&record.context, &record.input, cache) + .map_err(|source| ReactiveError::HandlerFailed { + handler_id: registered.id.clone(), + source, + })?; + + validate_effects(input_ref, &record.context, ®istered.id, &outcome.effects)?; + executions.push(HandlerExecution::from_outcome( + registered.id.clone(), + input_ref, + outcome, + )); + } + Ok(executions) + } + + fn dispatch_reports(&self, reports: &[Arc>]) { + for report in reports { + for hook in &self.hooks { + hook.on_report(report.clone()); + } + } + } +} + +#[derive(Clone, Debug)] +struct StorageFetchSlot { + address: Address, + slot: U256, + origins: Vec, +} + +#[derive(Clone, Debug)] +struct StorageFetchOrigin { + request_id: ResyncId, + target: ResyncTarget, +} + +#[derive(Clone, Debug)] +struct StorageFetchGroup { + block: ResyncBlock, + slots: Vec, + seen: HashSet<(Address, U256)>, +} + +fn execute_resync_requests(cache: &mut EvmCache, requests: &[ResyncRequest]) -> ResyncReport { + let mut failed = Vec::new(); + let mut storage_groups: Vec = Vec::new(); + + for request in requests { + for target in &request.targets { + match target { + ResyncTarget::StorageSlot { address, slot } => { + push_storage_resync_slot( + &mut storage_groups, + &request.id, + &request.block, + *address, + *slot, + ); + } + ResyncTarget::StorageSlots { address, slots } => { + for slot in slots { + push_storage_resync_slot( + &mut storage_groups, + &request.id, + &request.block, + *address, + *slot, + ); + } + } + ResyncTarget::Account { .. } => failed.push(ResyncFailure { + request_id: request.id.clone(), + block: request.block.clone(), + target: target.clone(), + kind: ResyncFailureKind::UnsupportedAccountTarget, + message: + "account resync is unsupported until a provider-neutral account fetcher exists" + .to_string(), + }), + } + } + } + + let mut state_updates = Vec::new(); + if !storage_groups.is_empty() { + if let Some(fetcher) = cache.storage_batch_fetcher().cloned() { + for group in storage_groups { + let block = group.block.clone(); + let fetches: Vec<(Address, U256)> = group + .slots + .iter() + .map(|slot| (slot.address, slot.slot)) + .collect(); + let results = (fetcher)(fetches, Some(resync_block_to_block_id(&block))); + let mut pending: HashMap<(Address, U256), StorageFetchSlot> = group + .slots + .iter() + .cloned() + .map(|slot| ((slot.address, slot.slot), slot)) + .collect(); + + for (address, slot, fetched) in results { + let Some(requested_slot) = pending.remove(&(address, slot)) else { + continue; + }; + match fetched { + Ok(value) => state_updates.push(StateUpdate::slot(address, slot, value)), + Err(error) => { + let message = error.to_string(); + push_resync_failures( + &mut failed, + &block, + requested_slot.origins, + ResyncFailureKind::StorageFetchFailed, + message, + ); + } + } + } + + for requested_slot in group.slots { + if pending + .remove(&(requested_slot.address, requested_slot.slot)) + .is_some() + { + push_resync_failures( + &mut failed, + &block, + requested_slot.origins, + ResyncFailureKind::StorageFetchOmitted, + "storage batch fetcher did not return a value for slot".to_string(), + ); + } + } + } + } else { + for group in storage_groups { + let block = group.block.clone(); + for slot in group.slots { + push_resync_failures( + &mut failed, + &block, + slot.origins, + ResyncFailureKind::MissingStorageFetcher, + "storage resync requires a storage batch fetcher".to_string(), + ); + } + } + } + } + + let diff = if state_updates.is_empty() { + StateDiff::default() + } else { + cache.apply_updates(&state_updates) + }; + + ResyncReport { + requested: requests.to_vec(), + state_updates, + diff, + failed, + } +} + +fn push_resync_failures( + failed: &mut Vec, + block: &ResyncBlock, + origins: Vec, + kind: ResyncFailureKind, + message: String, +) { + for origin in origins { + failed.push(ResyncFailure { + request_id: origin.request_id, + block: block.clone(), + target: origin.target, + kind, + message: message.clone(), + }); + } +} + +fn push_storage_resync_slot( + groups: &mut Vec, + request_id: &ResyncId, + block: &ResyncBlock, + address: Address, + slot: U256, +) { + let group_index = if let Some(index) = groups.iter().position(|group| group.block == *block) { + index + } else { + groups.push(StorageFetchGroup { + block: block.clone(), + slots: Vec::new(), + seen: HashSet::new(), + }); + groups.len() - 1 + }; + + let group = &mut groups[group_index]; + let origin = StorageFetchOrigin { + request_id: request_id.clone(), + target: ResyncTarget::StorageSlot { address, slot }, + }; + if group.seen.insert((address, slot)) { + group.slots.push(StorageFetchSlot { + address, + slot, + origins: vec![origin], + }); + } else if let Some(existing) = group + .slots + .iter_mut() + .find(|existing| existing.address == address && existing.slot == slot) + { + existing.origins.push(origin); + } +} + +fn resync_block_to_block_id(block: &ResyncBlock) -> BlockId { + match block { + ResyncBlock::Latest => BlockId::latest(), + ResyncBlock::Safe => BlockId::safe(), + ResyncBlock::Finalized => BlockId::finalized(), + ResyncBlock::Number(number) => BlockId::number(*number), + ResyncBlock::Hash { + number: _, + hash, + require_canonical, + } => BlockId::from((*hash, Some(*require_canonical))), + } +} + +impl RegisteredHandler { + fn matches(&self, input: &ReactiveInput) -> bool { + self.interests + .iter() + .any(|interest| interest_matches(interest, input)) + } + + fn route_log(&self, log: &Log) -> Option { + self.interests.iter().find_map(|interest| match interest { + ReactiveInterest::Logs(interest) if interest.matches(log) => Some(ReactiveLogRoute { + handler_id: self.id.clone(), + route_key: interest.route_key(log), + }), + ReactiveInterest::Logs(_) + | ReactiveInterest::Blocks(_) + | ReactiveInterest::PendingTransactions(_) => None, + }) + } +} + +fn merge_log_subscription_filter(filters: &mut Vec, next: &Filter) { + if let Some(existing) = filters + .iter_mut() + .find(|existing| existing.block_option == next.block_option) + { + merge_filter_set(&mut existing.address, &next.address); + for (existing_topic, next_topic) in existing.topics.iter_mut().zip(next.topics.iter()) { + merge_filter_set(existing_topic, next_topic); + } + } else { + filters.push(next.clone()); + } +} + +fn merge_filter_set(target: &mut FilterSet, source: &FilterSet) { + if target.is_empty() { + return; + } + if source.is_empty() { + *target = FilterSet::default(); + return; + } + for value in source.iter() { + target.insert(value.clone()); + } +} + +#[derive(Clone, Debug)] +struct HandlerExecution { + handler_id: HandlerId, + quality: StateEffectQuality, + tags: Vec, + state_updates: Vec, + invalidations: Vec, + resyncs: Vec, + speculative: Vec, + hook_signals: Vec, +} + +impl HandlerExecution { + fn from_outcome(handler_id: HandlerId, input_ref: InputRef, outcome: HandlerOutcome) -> Self { + let mut state_updates = Vec::new(); + let mut invalidations = Vec::new(); + let mut resyncs = Vec::new(); + let mut speculative = Vec::new(); + let mut hook_signals = Vec::new(); + + for effect in outcome.effects { + match effect { + ReactiveEffect::StateUpdate(update) => state_updates.push(update), + ReactiveEffect::Invalidate(invalidation) => { + state_updates.push(StateUpdate::purge( + invalidation.address, + invalidation.scope.clone(), + )); + invalidations.push(invalidation); + } + ReactiveEffect::Resync(request) => resyncs.push(request), + ReactiveEffect::Hook(signal) => hook_signals.push(signal), + ReactiveEffect::Speculative(mut request) => { + request.input_ref = input_ref; + speculative.push(request); + } + } + } + + Self { + handler_id, + quality: outcome.quality, + tags: outcome.tags, + state_updates, + invalidations, + resyncs, + speculative, + hook_signals, + } + } +} + +fn dedupe_records(records: Vec>) -> Vec> { + let mut seen = HashSet::new(); + let mut deduped = Vec::with_capacity(records.len()); + for record in records { + if seen.insert(record.input_ref()) { + deduped.push(record); + } + } + deduped +} + +fn sort_records(records: Vec>) -> Vec> { + let mut indexed: Vec<(usize, ReactiveInputRecord)> = + records.into_iter().enumerate().collect(); + indexed.sort_by_key(|(index, record)| record_sort_key(*index, record)); + indexed.into_iter().map(|(_, record)| record).collect() +} + +fn record_sort_key(index: usize, record: &ReactiveInputRecord) -> RecordSortKey { + if let ReactiveInput::Log(log) = &record.input + && is_canonical_status(&record.context.chain_status) + && !log.removed + { + return RecordSortKey { + class: 0, + block_number: log + .block_number + .or(record.context.block.as_ref().map(|block| block.number)) + .unwrap_or(u64::MAX), + transaction_index: log + .transaction_index + .or(record.context.transaction_index) + .unwrap_or(u64::MAX), + log_index: log + .log_index + .or(record.context.log_index) + .unwrap_or(u64::MAX), + original_index: index, + }; + } + + RecordSortKey { + class: 1, + block_number: 0, + transaction_index: 0, + log_index: 0, + original_index: index, + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +struct RecordSortKey { + class: u8, + block_number: u64, + transaction_index: u64, + log_index: u64, + original_index: usize, +} + +fn interest_matches(interest: &ReactiveInterest, input: &ReactiveInput) -> bool { + match (interest, input) { + (ReactiveInterest::Logs(interest), ReactiveInput::Log(log)) => interest.matches(log), + ( + ReactiveInterest::Blocks(BlockInterest { + mode: BlockInterestMode::Header, + }), + ReactiveInput::BlockHeader(_), + ) => true, + ( + ReactiveInterest::Blocks(BlockInterest { + mode: BlockInterestMode::FullBlock, + }), + ReactiveInput::FullBlock(_), + ) => true, + (ReactiveInterest::PendingTransactions(interest), ReactiveInput::PendingTxHash(_)) => { + interest.matches_hash_only() + } + (ReactiveInterest::PendingTransactions(interest), ReactiveInput::PendingTx(tx)) => { + interest.matches_tx(tx) + } + _ => false, + } +} + +fn validate_effects( + input_ref: InputRef, + ctx: &ReactiveContext, + handler_id: &HandlerId, + effects: &[ReactiveEffect], +) -> Result<(), ReactiveError> { + let pending = matches!(ctx.chain_status, ChainStatus::Pending) + || matches!(input_ref, InputRef::PendingTx { .. }); + if !pending { + return Ok(()); + } + + for effect in effects { + let effect_kind = match effect { + ReactiveEffect::StateUpdate(_) => Some("state_update"), + ReactiveEffect::Invalidate(_) => Some("invalidate"), + ReactiveEffect::Resync(_) => Some("resync"), + ReactiveEffect::Hook(_) | ReactiveEffect::Speculative(_) => None, + }; + if let Some(effect_kind) = effect_kind { + return Err(ReactiveError::InvalidPendingEffect { + input_ref: Box::new(input_ref), + handler_id: handler_id.clone(), + effect_kind, + }); + } + } + Ok(()) +} + +fn detect_conflicts( + input_ref: InputRef, + executions: &[HandlerExecution], +) -> Result<(), ReactiveError> { + let mut writes: HashMap = HashMap::new(); + for execution in executions { + for update in &execution.state_updates { + for (target, value) in absolute_writes(update) { + if let Some((previous_value, previous_handler)) = writes.get(&target) { + if previous_value != &value { + return Err(ReactiveError::ConflictingEffects { + input_ref: Box::new(input_ref), + target: Box::new(target), + first: previous_handler.clone(), + second: execution.handler_id.clone(), + }); + } + } else { + writes.insert(target, (value, execution.handler_id.clone())); + } + } + } + } + Ok(()) +} + +fn absolute_writes(update: &StateUpdate) -> Vec<(EffectTarget, AbsoluteValue)> { + match update { + StateUpdate::Slot { + address, + slot, + value, + } => vec![( + EffectTarget::StorageSlot { + address: *address, + slot: *slot, + }, + AbsoluteValue::U256(*value), + )], + StateUpdate::SlotMasked { + address, + slot, + mask, + value, + } => vec![( + EffectTarget::MaskedStorageSlot { + address: *address, + slot: *slot, + mask: *mask, + }, + AbsoluteValue::U256(*value), + )], + StateUpdate::Account { address, patch } | StateUpdate::AccountUpsert { address, patch } => { + account_patch_writes(*address, patch) + } + StateUpdate::SlotDelta { .. } + | StateUpdate::BalanceDelta { .. } + | StateUpdate::Purge { .. } => Vec::new(), + } +} + +fn account_patch_writes( + address: Address, + patch: &AccountPatch, +) -> Vec<(EffectTarget, AbsoluteValue)> { + let mut writes = Vec::new(); + if let Some(balance) = patch.balance { + writes.push(( + EffectTarget::AccountBalance { address }, + AbsoluteValue::U256(balance), + )); + } + if let Some(nonce) = patch.nonce { + writes.push(( + EffectTarget::AccountNonce { address }, + AbsoluteValue::U64(nonce), + )); + } + if let Some(code) = &patch.code { + writes.push(( + EffectTarget::AccountCode { address }, + AbsoluteValue::Bytes(code.clone()), + )); + } + writes +} + +fn input_ref(input: &ReactiveInput, ctx: &ReactiveContext) -> InputRef { + match input { + ReactiveInput::Log(log) => InputRef::Log { + chain_id: ctx.chain_id, + block_hash: log + .block_hash + .or(ctx.block.as_ref().map(|block| block.hash)) + .unwrap_or_default(), + transaction_hash: log.transaction_hash.unwrap_or_default(), + log_index: log.log_index.or(ctx.log_index).unwrap_or_default(), + }, + ReactiveInput::PendingTxHash(hash) => InputRef::PendingTx { + chain_id: ctx.chain_id, + hash: *hash, + }, + ReactiveInput::PendingTx(tx) => InputRef::PendingTx { + chain_id: ctx.chain_id, + hash: tx.tx_hash(), + }, + ReactiveInput::BlockHeader(header) => InputRef::Block { + chain_id: ctx.chain_id, + hash: header.hash(), + number: header.number(), + }, + ReactiveInput::FullBlock(block) => { + let header = block.header(); + InputRef::Block { + chain_id: ctx.chain_id, + hash: header.hash(), + number: header.number(), + } + } + } +} + +fn is_canonical_status(status: &ChainStatus) -> bool { + matches!( + status, + ChainStatus::Included { .. } | ChainStatus::Safe { .. } | ChainStatus::Finalized { .. } + ) +} + +/// Adapter that wraps a legacy [`EventDecoder`] as a log-only reactive handler. +pub struct EventDecoderHandler { + id: HandlerId, + decoder: Arc, + interest: LogInterest, +} + +impl EventDecoderHandler { + /// Create an adapter from a decoder and log interest. + pub fn new(id: HandlerId, decoder: Arc, interest: LogInterest) -> Self { + Self { + id, + decoder, + interest, + } + } +} + +impl ReactiveHandler for EventDecoderHandler { + fn id(&self) -> HandlerId { + self.id.clone() + } + + fn interests(&self) -> Vec> { + vec![ReactiveInterest::Logs(self.interest.clone())] + } + + fn handle( + &self, + _ctx: &ReactiveContext, + input: &ReactiveInput, + state: &dyn StateView, + ) -> Result { + let ReactiveInput::Log(log) = input else { + return Ok(HandlerOutcome::empty(StateEffectQuality::NoStateEffect)); + }; + + Ok(HandlerOutcome { + effects: self + .decoder + .decode(&log.inner, state) + .into_iter() + .map(ReactiveEffect::StateUpdate) + .collect(), + quality: StateEffectQuality::ExactFromInput, + tags: Vec::new(), + }) + } +} + +/// Provider-agnostic subscriber interface. +pub trait EventSubscriber: Send { + /// Register interests with the subscriber. + fn register_interests( + &mut self, + interests: &[ReactiveInterest], + ) -> Result<(), SubscriberError>; + + /// Return the next input batch, or `Ok(None)` when the stream is exhausted. + fn next_batch(&mut self) -> SubscriberNextBatch<'_, N>; +} + +/// Boxed future returned by [`EventSubscriber::next_batch`]. +pub type SubscriberNextBatch<'a, N> = Pin< + Box>, SubscriberError>> + Send + 'a>, +>; + +/// Subscriber mode requested for the Alloy scaffold. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)] +pub enum SubscriberMode { + /// Use provider pubsub streams when available. + #[default] + PubSub, + /// Use polling/watch APIs. + Polling, +} + +/// Subscriber configuration. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SubscriberConfig { + /// Hydrate pending transaction hashes into full bodies when possible. + pub hydrate_pending_transactions: bool, + /// Maximum records to emit per batch. + pub max_batch_size: usize, +} + +impl Default for SubscriberConfig { + fn default() -> Self { + Self { + hydrate_pending_transactions: false, + max_batch_size: 1024, + } + } +} + +/// Documented Alloy subscriber scaffold. +/// +/// 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. +pub struct AlloySubscriber { + provider: P, + mode: SubscriberMode, + config: SubscriberConfig, + interests: Vec>, + _network: PhantomData, +} + +impl AlloySubscriber { + /// Create a new Alloy subscriber scaffold. + pub fn new(provider: P, mode: SubscriberMode, config: SubscriberConfig) -> Self { + Self { + provider, + mode, + config, + interests: Vec::new(), + _network: PhantomData, + } + } + + /// Borrow the provider. + pub fn provider(&self) -> &P { + &self.provider + } + + /// Subscriber mode. + pub fn mode(&self) -> SubscriberMode { + self.mode + } + + /// Subscriber config. + pub fn config(&self) -> &SubscriberConfig { + &self.config + } + + /// Registered interests. + pub fn registered_interests(&self) -> &[ReactiveInterest] { + &self.interests + } +} + +impl EventSubscriber for AlloySubscriber { + fn register_interests( + &mut self, + interests: &[ReactiveInterest], + ) -> Result<(), SubscriberError> { + self.interests = interests.to_vec(); + Ok(()) + } + + fn next_batch(&mut self) -> SubscriberNextBatch<'_, N> { + Box::pin(async { + Err(SubscriberError::Unsupported( + "AlloySubscriber is a scaffold; live stream driving is not implemented in this feature slice", + )) + }) + } +} + +/// Subscriber error. +#[derive(Debug, thiserror::Error)] +pub enum SubscriberError { + /// Requested subscriber behavior is not implemented. + #[error("{0}")] + Unsupported(&'static str), + /// Provider or transport error. + #[error("provider error: {0}")] + Provider(String), +} diff --git a/tests/reactive_registry.rs b/tests/reactive_registry.rs new file mode 100644 index 0000000..509d40d --- /dev/null +++ b/tests/reactive_registry.rs @@ -0,0 +1,75 @@ +//! Implementation-owned registry regression tests. +#![cfg(feature = "reactive")] + +use std::sync::Arc; + +use alloy_network::Ethereum; +use alloy_primitives::Address; +use alloy_rpc_types_eth::Filter; + +use evm_fork_cache::events::StateView; +use evm_fork_cache::reactive::{ + HandlerError, HandlerId, HandlerOutcome, LogInterest, ReactiveContext, ReactiveHandler, + ReactiveInput, ReactiveInterest, ReactiveRegistry, RegisterError, RouteKeySpec, + StateEffectQuality, +}; + +struct NoopHandler { + id: HandlerId, + address: Address, +} + +impl NoopHandler { + fn new(id: &'static str, address: Address) -> Self { + Self { + id: HandlerId::new(id), + address, + } + } +} + +impl ReactiveHandler for NoopHandler { + fn id(&self) -> HandlerId { + self.id.clone() + } + + fn interests(&self) -> Vec { + vec![ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(self.address), + local_matcher: None, + route_key: Some(RouteKeySpec::EmitterAddress), + })] + } + + fn handle( + &self, + _ctx: &ReactiveContext, + _input: &ReactiveInput, + _state: &dyn StateView, + ) -> Result { + Ok(HandlerOutcome::empty(StateEffectQuality::NoStateEffect)) + } +} + +#[test] +fn reactive_registry_rejects_duplicate_handler_ids() { + let mut registry = ReactiveRegistry::::new(); + registry + .register_handler(Arc::new(NoopHandler::new( + "duplicate", + Address::repeat_byte(0x11), + ))) + .expect("first registration should succeed"); + + let err = registry + .register_handler(Arc::new(NoopHandler::new( + "duplicate", + Address::repeat_byte(0x22), + ))) + .expect_err("duplicate handler id must be rejected"); + + assert!(matches!( + err, + RegisterError::DuplicateHandler(id) if id == HandlerId::new("duplicate") + )); +} diff --git a/tests/reactive_resync.rs b/tests/reactive_resync.rs new file mode 100644 index 0000000..bba1918 --- /dev/null +++ b/tests/reactive_resync.rs @@ -0,0 +1,444 @@ +//! Manager-authored acceptance tests for reactive resync execution. +//! +//! These tests pin the next runtime slice after routing: handlers can already +//! emit `ResyncRequest`s, but the runtime must also be able to execute storage +//! resyncs after direct effects, apply authoritative values through +//! `StateUpdate`, and report both applied and failed resync targets. +#![cfg(feature = "reactive")] + +mod common; + +use std::sync::{Arc, Mutex}; + +use alloy_eips::BlockId; +use alloy_network::Ethereum; +use alloy_primitives::{Address, B256, Bytes, Log as PrimitiveLog, U256, keccak256}; +use alloy_rpc_types_eth::{Filter, Log}; +use anyhow::{Result, anyhow}; + +use common::setup_cache; +use evm_fork_cache::StateUpdate; +use evm_fork_cache::events::StateView; +use evm_fork_cache::reactive::{ + AccountFieldMask, BlockRef, ChainStatus, HandlerError, HandlerId, HandlerOutcome, InputSource, + LogInterest, ReactiveConfig, ReactiveContext, ReactiveEffect, ReactiveHandler, ReactiveInput, + ReactiveInputBatch, ReactiveInputRecord, ReactiveInterest, ReactiveReport, ReactiveRuntime, + ResyncBlock, ResyncFailureKind, ResyncId, ResyncPriority, ResyncReason, ResyncRequest, + ResyncTarget, RouteKeySpec, StateEffectQuality, +}; + +fn rpc_log(address: Address, topics: Vec, block_number: u64) -> Log { + Log { + inner: PrimitiveLog::new_unchecked(address, topics, 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(0x44)), + transaction_index: Some(0), + log_index: Some(0), + removed: false, + } +} + +fn included_context(block_number: u64) -> ReactiveContext { + let block = BlockRef { + number: block_number, + hash: B256::repeat_byte(block_number as u8), + parent_hash: Some(B256::repeat_byte(block_number.saturating_sub(1) as u8)), + timestamp: Some(1_700_000_000 + block_number), + }; + + ReactiveContext { + chain_id: Some(1), + source: InputSource::Batch, + chain_status: ChainStatus::Included { + block: block.clone(), + confirmations: 0, + }, + block: Some(block), + transaction_index: Some(0), + log_index: Some(0), + } +} + +fn batch(input: ReactiveInput, ctx: ReactiveContext) -> ReactiveInputBatch { + ReactiveInputBatch::new(vec![ReactiveInputRecord::new(input, ctx)]) +} + +struct WriteThenResync { + address: Address, + slot: U256, + block_hash: B256, +} + +impl ReactiveHandler for WriteThenResync { + fn id(&self) -> HandlerId { + HandlerId::new("write-then-resync") + } + + fn interests(&self) -> Vec { + vec![ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(self.address), + local_matcher: None, + route_key: Some(RouteKeySpec::EmitterAddress), + })] + } + + fn handle( + &self, + _ctx: &ReactiveContext, + _input: &ReactiveInput, + _state: &dyn StateView, + ) -> Result { + Ok(HandlerOutcome { + effects: vec![ + ReactiveEffect::StateUpdate(StateUpdate::slot( + self.address, + self.slot, + U256::from(1), + )), + ReactiveEffect::Resync(ResyncRequest { + id: ResyncId::new("slot-repair"), + reason: ResyncReason::HandlerRequested, + block: ResyncBlock::Hash { + number: 60, + hash: self.block_hash, + require_canonical: true, + }, + targets: vec![ResyncTarget::StorageSlot { + address: self.address, + slot: self.slot, + }], + priority: ResyncPriority::High, + }), + ], + quality: StateEffectQuality::AppliedWithPendingResync, + tags: vec![], + }) + } +} + +struct MixedResyncTargets { + address: Address, + slot_a: U256, + slot_b: U256, +} + +impl ReactiveHandler for MixedResyncTargets { + fn id(&self) -> HandlerId { + HandlerId::new("mixed-resync") + } + + fn interests(&self) -> Vec { + vec![ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(self.address), + local_matcher: None, + route_key: Some(RouteKeySpec::EmitterAddress), + })] + } + + fn handle( + &self, + _ctx: &ReactiveContext, + _input: &ReactiveInput, + _state: &dyn StateView, + ) -> Result { + Ok(HandlerOutcome { + effects: vec![ReactiveEffect::Resync(ResyncRequest { + id: ResyncId::new("mixed-targets"), + reason: ResyncReason::HandlerRequested, + block: ResyncBlock::Number(61), + targets: vec![ + ResyncTarget::StorageSlots { + address: self.address, + slots: vec![self.slot_a, self.slot_b], + }, + ResyncTarget::Account { + address: self.address, + fields: AccountFieldMask { + balance: true, + nonce: false, + code: false, + }, + }, + ], + priority: ResyncPriority::Normal, + })], + quality: StateEffectQuality::AppliedWithPendingResync, + tags: vec![], + }) + } +} + +struct DuplicateSlotResyncs { + address: Address, + slot: U256, +} + +impl ReactiveHandler for DuplicateSlotResyncs { + fn id(&self) -> HandlerId { + HandlerId::new("duplicate-slot-resyncs") + } + + fn interests(&self) -> Vec { + vec![ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(self.address), + local_matcher: None, + route_key: Some(RouteKeySpec::EmitterAddress), + })] + } + + fn handle( + &self, + _ctx: &ReactiveContext, + _input: &ReactiveInput, + _state: &dyn StateView, + ) -> Result { + Ok(HandlerOutcome { + effects: vec![ + ReactiveEffect::Resync(ResyncRequest { + id: ResyncId::new("duplicate-a"), + reason: ResyncReason::HandlerRequested, + block: ResyncBlock::Number(62), + targets: vec![ResyncTarget::StorageSlot { + address: self.address, + slot: self.slot, + }], + priority: ResyncPriority::Normal, + }), + ReactiveEffect::Resync(ResyncRequest { + id: ResyncId::new("duplicate-b"), + reason: ResyncReason::HandlerRequested, + block: ResyncBlock::Number(62), + targets: vec![ResyncTarget::StorageSlots { + address: self.address, + slots: vec![self.slot], + }], + priority: ResyncPriority::Normal, + }), + ], + quality: StateEffectQuality::AppliedWithPendingResync, + tags: vec![], + }) + } +} + +#[tokio::test] +async fn reactive_runtime_executes_storage_resync_after_direct_effects() -> Result<()> { + let address = Address::repeat_byte(0x91); + let slot = U256::from(3); + let block_hash = B256::repeat_byte(0x60); + let seen_fetches = Arc::new(Mutex::new(Vec::new())); + let mut cache = setup_cache().await?; + cache.set_storage_batch_fetcher({ + let seen_fetches = seen_fetches.clone(); + Arc::new(move |requests, block| { + seen_fetches.lock().unwrap().push((requests.clone(), block)); + requests + .into_iter() + .map(|(addr, slot)| (addr, slot, Ok(U256::from(42)))) + .collect() + }) + }); + + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(WriteThenResync { + address, + slot, + block_hash, + }))?; + + let report = runtime.ingest_batch_with_resync( + &mut cache, + batch( + ReactiveInput::Log(rpc_log(address, vec![keccak256(b"Repair()")], 60)), + included_context(60), + ), + )?; + + assert_eq!( + cache.cached_storage_value(address, slot), + Some(U256::from(42)), + "authoritative resync value must overwrite direct handler effects" + ); + + let fetches = seen_fetches.lock().unwrap(); + assert_eq!(fetches.len(), 1); + assert_eq!(fetches[0].0, vec![(address, slot)]); + match &fetches[0].1 { + Some(BlockId::Hash(hash)) => { + assert_eq!(hash.block_hash, block_hash); + assert_eq!(hash.require_canonical, Some(true)); + } + other => panic!("expected canonical block hash fetch, got {other:?}"), + } + + let resynced: Vec<_> = report + .reports + .iter() + .filter_map(|report| match report.as_ref() { + ReactiveReport::Resynced(report) => Some(report), + _ => None, + }) + .collect(); + assert_eq!(resynced.len(), 1); + assert_eq!(resynced[0].requested.len(), 1); + assert_eq!(resynced[0].state_updates.len(), 1); + assert_eq!(resynced[0].diff.slots.len(), 1); + assert_eq!(resynced[0].diff.slots[0].old, U256::from(1)); + assert_eq!(resynced[0].diff.slots[0].new, U256::from(42)); + assert!(resynced[0].failed.is_empty()); + + Ok(()) +} + +#[tokio::test] +async fn reactive_runtime_batches_resync_slots_and_reports_failed_targets() -> Result<()> { + let address = Address::repeat_byte(0x92); + let slot_a = U256::from(10); + let slot_b = U256::from(11); + let seen_fetches = Arc::new(Mutex::new(Vec::new())); + let mut cache = setup_cache().await?; + cache.set_storage_batch_fetcher({ + let seen_fetches = seen_fetches.clone(); + Arc::new(move |requests, block| { + seen_fetches.lock().unwrap().push((requests.clone(), block)); + requests + .into_iter() + .map(|(addr, slot)| { + if slot == slot_b { + (addr, slot, Err(anyhow!("stub slot failure"))) + } else { + (addr, slot, Ok(U256::from(777))) + } + }) + .collect() + }) + }); + + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(MixedResyncTargets { + address, + slot_a, + slot_b, + }))?; + + let report = runtime.ingest_batch_with_resync( + &mut cache, + batch( + ReactiveInput::Log(rpc_log(address, vec![keccak256(b"MixedRepair()")], 61)), + included_context(61), + ), + )?; + + assert_eq!( + cache.cached_storage_value(address, slot_a), + Some(U256::from(777)) + ); + assert_eq!(cache.cached_storage_value(address, slot_b), None); + + let fetches = seen_fetches.lock().unwrap(); + assert_eq!(fetches.len(), 1); + assert_eq!(fetches[0].0, vec![(address, slot_a), (address, slot_b)]); + assert_eq!(fetches[0].1, Some(BlockId::number(61))); + + let resynced: Vec<_> = report + .reports + .iter() + .filter_map(|report| match report.as_ref() { + ReactiveReport::Resynced(report) => Some(report), + _ => None, + }) + .collect(); + assert_eq!(resynced.len(), 1); + assert_eq!(resynced[0].requested.len(), 1); + assert_eq!(resynced[0].state_updates.len(), 1); + assert_eq!( + resynced[0].state_updates[0], + StateUpdate::slot(address, slot_a, U256::from(777)) + ); + assert_eq!(resynced[0].diff.slots.len(), 1); + assert_eq!(resynced[0].failed.len(), 2); + assert!(resynced[0].failed.iter().any(|failure| matches!( + failure.target, + ResyncTarget::StorageSlot { address: failed_address, slot: failed_slot } + if failed_address == address && failed_slot == slot_b + ) && failure.kind + == ResyncFailureKind::StorageFetchFailed + && failure.message.contains("stub slot failure"))); + assert!(resynced[0].failed.iter().any(|failure| matches!( + failure.target, + ResyncTarget::Account { .. } + ) && failure.kind + == ResyncFailureKind::UnsupportedAccountTarget + && failure.message.contains("account"))); + + Ok(()) +} + +#[tokio::test] +async fn reactive_runtime_fans_out_duplicate_resync_failures_to_all_request_origins() -> Result<()> +{ + let address = Address::repeat_byte(0x93); + let slot = U256::from(12); + let seen_fetches = Arc::new(Mutex::new(Vec::new())); + let mut cache = setup_cache().await?; + cache.set_storage_batch_fetcher({ + let seen_fetches = seen_fetches.clone(); + Arc::new(move |requests, block| { + seen_fetches.lock().unwrap().push((requests.clone(), block)); + requests + .into_iter() + .map(|(addr, slot)| (addr, slot, Err(anyhow!("shared fetch failure")))) + .collect() + }) + }); + + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(DuplicateSlotResyncs { address, slot }))?; + + let report = runtime.ingest_batch_with_resync( + &mut cache, + batch( + ReactiveInput::Log(rpc_log(address, vec![keccak256(b"DuplicateRepair()")], 62)), + included_context(62), + ), + )?; + + let fetches = seen_fetches.lock().unwrap(); + assert_eq!( + fetches.len(), + 1, + "duplicate storage targets should share one provider fetch" + ); + assert_eq!(fetches[0].0, vec![(address, slot)]); + + let resynced: Vec<_> = report + .reports + .iter() + .filter_map(|report| match report.as_ref() { + ReactiveReport::Resynced(report) => Some(report), + _ => None, + }) + .collect(); + assert_eq!(resynced.len(), 1); + assert!(resynced[0].state_updates.is_empty()); + assert_eq!( + resynced[0].failed.len(), + 2, + "each originating request must get an explicit failure" + ); + let failed_ids: Vec<_> = resynced[0] + .failed + .iter() + .map(|failure| failure.request_id.clone()) + .collect(); + assert!(failed_ids.contains(&ResyncId::new("duplicate-a"))); + assert!(failed_ids.contains(&ResyncId::new("duplicate-b"))); + assert!(resynced[0].failed.iter().all(|failure| { + failure.kind == ResyncFailureKind::StorageFetchFailed + && failure.message.contains("shared fetch failure") + })); + + Ok(()) +} diff --git a/tests/reactive_router.rs b/tests/reactive_router.rs new file mode 100644 index 0000000..9d2e2f5 --- /dev/null +++ b/tests/reactive_router.rs @@ -0,0 +1,224 @@ +//! Manager-authored acceptance tests for reactive routing and filter planning. +//! +//! These tests pin the public behavior for the provider filter consolidation +//! and routing-index phase. They should fail before the router/registry surface +//! exists and pass once filters are consolidated as safe supersets while local +//! routing remains exact. +#![cfg(feature = "reactive")] + +use std::sync::Arc; + +use alloy_network::Ethereum; +use alloy_primitives::{Address, B256, Bytes, Log as PrimitiveLog, keccak256}; +use alloy_rpc_types_eth::{Filter, Log}; +use anyhow::Result; + +use evm_fork_cache::events::StateView; +use evm_fork_cache::reactive::{ + HandlerError, HandlerId, HandlerOutcome, LogInterest, LogMatcher, ReactiveContext, + ReactiveEffect, ReactiveHandler, ReactiveInput, ReactiveInterest, ReactiveRegistry, RouteKey, + RouteKeySpec, StateEffectQuality, +}; + +fn rpc_log(address: Address, topics: Vec) -> Log { + Log { + inner: PrimitiveLog::new_unchecked(address, topics, Bytes::new()), + block_hash: Some(B256::repeat_byte(0x10)), + block_number: Some(10), + block_timestamp: Some(1_700_000_010), + transaction_hash: Some(B256::repeat_byte(0x20)), + transaction_index: Some(0), + log_index: Some(0), + removed: false, + } +} + +#[derive(Clone)] +struct NoopHandler { + id: HandlerId, + interest: LogInterest, +} + +impl NoopHandler { + fn new(id: &'static str, interest: LogInterest) -> Self { + Self { + id: HandlerId::new(id), + interest, + } + } +} + +impl ReactiveHandler for NoopHandler { + fn id(&self) -> HandlerId { + self.id.clone() + } + + fn interests(&self) -> Vec { + vec![ReactiveInterest::Logs(self.interest.clone())] + } + + fn handle( + &self, + _ctx: &ReactiveContext, + _input: &ReactiveInput, + _state: &dyn StateView, + ) -> Result { + Ok(HandlerOutcome { + effects: vec![ReactiveEffect::Hook(evm_fork_cache::reactive::HookSignal { + namespace: "test".into(), + kind: self.id.as_str().to_owned().into(), + labels: vec![], + payload: None, + })], + quality: StateEffectQuality::NoStateEffect, + tags: vec![], + }) + } +} + +struct TopicMatcher { + index: usize, + value: B256, +} + +impl LogMatcher for TopicMatcher { + fn matches(&self, log: &Log) -> bool { + log.topics().get(self.index) == Some(&self.value) + } +} + +#[test] +fn reactive_registry_consolidates_provider_filters_as_safe_superset() -> Result<()> { + let token_a = Address::repeat_byte(0xa1); + let token_b = Address::repeat_byte(0xb2); + let sig_a = keccak256(b"TokenAEvent()"); + let sig_b = keccak256(b"TokenBEvent()"); + + let mut registry = ReactiveRegistry::::new(); + registry.register_handler(Arc::new(NoopHandler::new( + "token-a", + LogInterest { + provider_filter: Filter::new().address(token_a).event_signature(sig_a), + local_matcher: None, + route_key: Some(RouteKeySpec::EmitterAddress), + }, + )))?; + registry.register_handler(Arc::new(NoopHandler::new( + "token-b", + LogInterest { + provider_filter: Filter::new().address(token_b).event_signature(sig_b), + local_matcher: None, + route_key: Some(RouteKeySpec::EmitterAddress), + }, + )))?; + + let filters = registry.log_subscription_filters(); + assert_eq!(filters.len(), 1, "compatible log interests should merge"); + let consolidated = &filters[0]; + + let wanted_a = rpc_log(token_a, vec![sig_a]); + let wanted_b = rpc_log(token_b, vec![sig_b]); + let overfetched = rpc_log(token_a, vec![sig_b]); + + assert!(consolidated.rpc_matches(&wanted_a)); + assert!(consolidated.rpc_matches(&wanted_b)); + assert!( + consolidated.rpc_matches(&overfetched), + "merged filters may be a safe provider-side superset" + ); + + let route_a = registry.route_log(&wanted_a); + let route_b = registry.route_log(&wanted_b); + let overfetch_routes = registry.route_log(&overfetched); + + assert_eq!(route_a.len(), 1); + assert_eq!(route_a[0].handler_id, HandlerId::new("token-a")); + assert_eq!(route_a[0].route_key, Some(RouteKey::Address(token_a))); + assert_eq!(route_b.len(), 1); + assert_eq!(route_b[0].handler_id, HandlerId::new("token-b")); + assert_eq!(route_b[0].route_key, Some(RouteKey::Address(token_b))); + assert!( + overfetch_routes.is_empty(), + "local routing must remain exact after provider-side consolidation" + ); + + Ok(()) +} + +#[test] +fn reactive_router_routes_shared_emitters_by_route_key_in_registration_order() -> Result<()> { + let vault = Address::repeat_byte(0xcc); + let swap_sig = keccak256(b"Swap(bytes32,address,int256)"); + let pool_a = B256::repeat_byte(0xa0); + let pool_b = B256::repeat_byte(0xb0); + let pool_c = B256::repeat_byte(0xc0); + + let all_swaps = NoopHandler::new( + "all-swaps", + LogInterest { + provider_filter: Filter::new().address(vault).event_signature(swap_sig), + local_matcher: None, + route_key: Some(RouteKeySpec::EmitterAddress), + }, + ); + let pool_a_handler = NoopHandler::new( + "pool-a", + LogInterest { + provider_filter: Filter::new().address(vault).event_signature(swap_sig), + local_matcher: Some(Arc::new(TopicMatcher { + index: 1, + value: pool_a, + })), + route_key: Some(RouteKeySpec::Topic { index: 1 }), + }, + ); + let pool_b_handler = NoopHandler::new( + "pool-b", + LogInterest { + provider_filter: Filter::new().address(vault).event_signature(swap_sig), + local_matcher: Some(Arc::new(TopicMatcher { + index: 1, + value: pool_b, + })), + route_key: Some(RouteKeySpec::Topic { index: 1 }), + }, + ); + + let mut registry = ReactiveRegistry::::new(); + registry.register_handler(Arc::new(all_swaps))?; + registry.register_handler(Arc::new(pool_a_handler))?; + registry.register_handler(Arc::new(pool_b_handler))?; + + let filters = registry.log_subscription_filters(); + assert_eq!( + filters.len(), + 1, + "shared-emitter interests should share one provider filter" + ); + assert!(filters[0].rpc_matches(&rpc_log(vault, vec![swap_sig, pool_a]))); + assert!(filters[0].rpc_matches(&rpc_log(vault, vec![swap_sig, pool_b]))); + + let routes = registry.route_log(&rpc_log(vault, vec![swap_sig, pool_b])); + let routed: Vec<_> = routes + .iter() + .map(|route| (route.handler_id.clone(), route.route_key.clone())) + .collect(); + assert_eq!( + routed, + vec![ + (HandlerId::new("all-swaps"), Some(RouteKey::Address(vault))), + (HandlerId::new("pool-b"), Some(RouteKey::Bytes32(pool_b))), + ], + "matching handlers must be returned in registration order with route keys" + ); + + let unrelated_pool_routes = registry.route_log(&rpc_log(vault, vec![swap_sig, pool_c])); + assert_eq!(unrelated_pool_routes.len(), 1); + assert_eq!( + unrelated_pool_routes[0].handler_id, + HandlerId::new("all-swaps"), + "custom local matchers must exclude nonmatching shared-emitter handlers" + ); + + Ok(()) +} diff --git a/tests/reactive_runtime.rs b/tests/reactive_runtime.rs new file mode 100644 index 0000000..6fdf74e --- /dev/null +++ b/tests/reactive_runtime.rs @@ -0,0 +1,609 @@ +//! Manager-authored acceptance tests for the reactive runtime feature. +//! +//! These tests intentionally describe the new public contract before the +//! implementation exists. They should fail on the current log-only event pipeline +//! and pass once `evm_fork_cache::reactive` provides a provider-agnostic handler +//! runtime. +#![cfg(feature = "reactive")] + +mod common; + +use std::sync::{Arc, Mutex}; + +use alloy_network::Ethereum; +use alloy_primitives::{Address, B256, Bytes, Log as PrimitiveLog, U256, keccak256}; +use alloy_rpc_types_eth::{Filter, Log}; +use anyhow::Result; + +use common::{install_mock_erc20, setup_cache}; +use evm_fork_cache::events::StateView; +use evm_fork_cache::reactive::{ + AppliedReport, BlockRef, ChainStatus, HandlerError, HandlerId, HandlerOutcome, HookSignal, + InputRef, InputSource, InvalidationReason, InvalidationRequest, LogInterest, LogMatcher, + PendingTxInterest, ReactiveConfig, ReactiveContext, ReactiveEffect, ReactiveError, + ReactiveHandler, ReactiveHook, ReactiveInput, ReactiveInputBatch, ReactiveInputRecord, + ReactiveInterest, ReactiveReport, ReactiveRuntime, ReportTag, ResyncBlock, ResyncId, + ResyncPriority, ResyncReason, ResyncRequest, ResyncTarget, RouteKeySpec, SpeculativeId, + SpeculativeRequest, StateEffectQuality, +}; +use evm_fork_cache::{PurgeScope, StateUpdate}; + +fn rpc_log( + address: Address, + topics: Vec, + block_number: u64, + tx_index: u64, + log_index: u64, +) -> Log { + let block_hash = B256::repeat_byte(block_number as u8); + Log { + inner: PrimitiveLog::new_unchecked(address, topics, Bytes::new()), + block_hash: Some(block_hash), + block_number: Some(block_number), + block_timestamp: Some(1_700_000_000 + block_number), + transaction_hash: Some(B256::repeat_byte((tx_index + 1) as u8)), + transaction_index: Some(tx_index), + log_index: Some(log_index), + removed: false, + } +} + +fn included_context(block_number: u64, log_index: u64) -> ReactiveContext { + let block = BlockRef { + number: block_number, + hash: B256::repeat_byte(block_number as u8), + parent_hash: Some(B256::repeat_byte(block_number.saturating_sub(1) as u8)), + timestamp: Some(1_700_000_000 + block_number), + }; + + ReactiveContext { + chain_id: Some(1), + source: InputSource::Batch, + chain_status: ChainStatus::Included { + block: block.clone(), + confirmations: 0, + }, + block: Some(block), + transaction_index: Some(0), + log_index: Some(log_index), + } +} + +fn batch(records: Vec<(ReactiveInput, ReactiveContext)>) -> ReactiveInputBatch { + ReactiveInputBatch::new( + records + .into_iter() + .map(|(input, ctx)| ReactiveInputRecord::new(input, ctx)) + .collect(), + ) +} + +#[derive(Clone)] +struct SlotWriter { + id: HandlerId, + interest: ReactiveInterest, + slot: U256, + value: U256, + hook_kind: &'static str, +} + +impl SlotWriter { + fn any_log_from(id: &'static str, address: Address, slot: U256, value: U256) -> Self { + Self { + id: HandlerId::new(id), + interest: ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(address), + local_matcher: None, + route_key: Some(RouteKeySpec::EmitterAddress), + }), + slot, + value, + hook_kind: "slot.write", + } + } +} + +impl ReactiveHandler for SlotWriter { + fn id(&self) -> HandlerId { + self.id.clone() + } + + fn interests(&self) -> Vec { + vec![self.interest.clone()] + } + + fn handle( + &self, + _ctx: &ReactiveContext, + input: &ReactiveInput, + _state: &dyn StateView, + ) -> Result { + let ReactiveInput::Log(log) = input else { + return Ok(HandlerOutcome::empty(StateEffectQuality::NoStateEffect)); + }; + + Ok(HandlerOutcome { + effects: vec![ + ReactiveEffect::StateUpdate(StateUpdate::slot( + log.address(), + self.slot, + self.value, + )), + ReactiveEffect::Hook(HookSignal { + namespace: "test".into(), + kind: self.hook_kind.into(), + labels: vec![ReportTag::new("handler", self.id.as_str())], + payload: None, + }), + ], + quality: StateEffectQuality::ExactFromInput, + tags: vec![ReportTag::new("slot", self.slot.to_string())], + }) + } +} + +struct LogIndexWriter { + id: HandlerId, + address: Address, + slot: U256, +} + +impl ReactiveHandler for LogIndexWriter { + fn id(&self) -> HandlerId { + self.id.clone() + } + + fn interests(&self) -> Vec { + vec![ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(self.address), + local_matcher: None, + route_key: Some(RouteKeySpec::EmitterAddress), + })] + } + + fn handle( + &self, + ctx: &ReactiveContext, + input: &ReactiveInput, + _state: &dyn StateView, + ) -> Result { + let ReactiveInput::Log(log) = input else { + return Ok(HandlerOutcome::empty(StateEffectQuality::NoStateEffect)); + }; + let value = U256::from(ctx.log_index.expect("log context carries log index")); + Ok(HandlerOutcome { + effects: vec![ReactiveEffect::StateUpdate(StateUpdate::slot( + log.address(), + self.slot, + value, + ))], + quality: StateEffectQuality::ExactFromInput, + tags: vec![], + }) + } +} + +struct TopicMatcher { + index: usize, + value: B256, +} + +impl LogMatcher for TopicMatcher { + fn matches(&self, log: &Log) -> bool { + log.topics().get(self.index) == Some(&self.value) + } +} + +#[derive(Default)] +struct RecordingHook { + applied_values: Arc>>, + hook_signals: Arc>>, +} + +impl ReactiveHook for RecordingHook { + fn on_report(&self, report: Arc>) { + if let ReactiveReport::Applied(AppliedReport { + diff, hook_signals, .. + }) = report.as_ref() + { + self.applied_values + .lock() + .unwrap() + .extend(diff.slots.iter().map(|change| change.new)); + self.hook_signals.lock().unwrap().extend( + hook_signals + .iter() + .map(|signal| format!("{}:{}", signal.namespace, signal.kind)), + ); + } + } +} + +#[tokio::test] +async fn reactive_runtime_applies_state_updates_and_dispatches_applied_hooks() -> Result<()> { + let emitter = Address::repeat_byte(0x71); + let slot = U256::from(7); + let value = U256::from(99); + let mut cache = setup_cache().await?; + install_mock_erc20(&mut cache, emitter); + + let hook = Arc::new(RecordingHook::default()); + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(SlotWriter::any_log_from( + "writer", emitter, slot, value, + )))?; + runtime.register_hook(hook.clone())?; + + let report = runtime.ingest_batch( + &mut cache, + batch(vec![( + ReactiveInput::Log(rpc_log(emitter, vec![keccak256(b"Write()")], 10, 0, 0)), + included_context(10, 0), + )]), + )?; + + assert_eq!(cache.cached_storage_value(emitter, slot), Some(value)); + assert_eq!(report.applied.len(), 1); + assert_eq!(report.applied[0].handler_id, HandlerId::new("writer")); + assert_eq!( + report.applied[0].quality, + StateEffectQuality::ExactFromInput + ); + assert_eq!(*hook.applied_values.lock().unwrap(), vec![value]); + assert_eq!( + *hook.hook_signals.lock().unwrap(), + vec!["test:slot.write".to_string()] + ); + Ok(()) +} + +#[tokio::test] +async fn reactive_runtime_orders_logs_dedupes_inputs_and_allows_sequential_writes() -> Result<()> { + let emitter = Address::repeat_byte(0x72); + let slot = U256::from(8); + let mut cache = setup_cache().await?; + install_mock_erc20(&mut cache, emitter); + + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(LogIndexWriter { + id: HandlerId::new("log-index-writer"), + address: emitter, + slot, + }))?; + + let topic = keccak256(b"Ordered()"); + let block = 20; + let report = runtime.ingest_batch( + &mut cache, + batch(vec![ + ( + ReactiveInput::Log(rpc_log(emitter, vec![topic], block, 0, 2)), + included_context(block, 2), + ), + ( + ReactiveInput::Log(rpc_log(emitter, vec![topic], block, 0, 1)), + included_context(block, 1), + ), + ( + ReactiveInput::Log(rpc_log(emitter, vec![topic], block, 0, 1)), + included_context(block, 1), + ), + ]), + )?; + + let applied_log_indexes: Vec = report + .applied + .iter() + .map(|applied| match applied.input_ref { + InputRef::Log { log_index, .. } => log_index, + _ => panic!("expected log input ref"), + }) + .collect(); + + assert_eq!(applied_log_indexes, vec![1, 2]); + assert_eq!( + cache.cached_storage_value(emitter, slot), + Some(U256::from(2)) + ); + Ok(()) +} + +#[tokio::test] +async fn reactive_runtime_routes_shared_emitters_with_local_topic_matchers() -> Result<()> { + let vault = Address::repeat_byte(0x73); + let pool_a = B256::repeat_byte(0xa0); + let pool_b = B256::repeat_byte(0xb0); + let slot = U256::from(9); + let mut cache = setup_cache().await?; + install_mock_erc20(&mut cache, vault); + + let mut handler_a = SlotWriter::any_log_from("pool-a", vault, slot, U256::from(100)); + handler_a.interest = ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(vault), + local_matcher: Some(Arc::new(TopicMatcher { + index: 1, + value: pool_a, + })), + route_key: Some(RouteKeySpec::Topic { index: 1 }), + }); + + let mut handler_b = SlotWriter::any_log_from("pool-b", vault, slot, U256::from(200)); + handler_b.interest = ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(vault), + local_matcher: Some(Arc::new(TopicMatcher { + index: 1, + value: pool_b, + })), + route_key: Some(RouteKeySpec::Topic { index: 1 }), + }); + + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(handler_a))?; + runtime.register_handler(Arc::new(handler_b))?; + + let report = runtime.ingest_batch( + &mut cache, + batch(vec![( + ReactiveInput::Log(rpc_log( + vault, + vec![keccak256(b"Swap(bytes32)"), pool_b], + 30, + 0, + 0, + )), + included_context(30, 0), + )]), + )?; + + assert_eq!(report.applied.len(), 1); + assert_eq!(report.applied[0].handler_id, HandlerId::new("pool-b")); + assert_eq!( + cache.cached_storage_value(vault, slot), + Some(U256::from(200)) + ); + Ok(()) +} + +#[tokio::test] +async fn reactive_runtime_rejects_conflicting_effects_for_one_input() -> Result<()> { + let emitter = Address::repeat_byte(0x74); + let slot = U256::from(10); + let mut cache = setup_cache().await?; + let before = cache.cached_storage_value(emitter, slot); + + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(SlotWriter::any_log_from( + "first", + emitter, + slot, + U256::from(1), + )))?; + runtime.register_handler(Arc::new(SlotWriter::any_log_from( + "second", + emitter, + slot, + U256::from(2), + )))?; + + let err = runtime + .ingest_batch( + &mut cache, + batch(vec![( + ReactiveInput::Log(rpc_log(emitter, vec![keccak256(b"Conflict()")], 40, 0, 0)), + included_context(40, 0), + )]), + ) + .expect_err("conflicting writes must be rejected before mutation"); + + assert!(matches!(err, ReactiveError::ConflictingEffects { .. })); + assert_eq!(cache.cached_storage_value(emitter, slot), before); + Ok(()) +} + +struct PendingCanonicalWriter; + +impl ReactiveHandler for PendingCanonicalWriter { + fn id(&self) -> HandlerId { + HandlerId::new("pending-writer") + } + + fn interests(&self) -> Vec { + vec![ReactiveInterest::PendingTransactions( + PendingTxInterest::default(), + )] + } + + fn handle( + &self, + _ctx: &ReactiveContext, + _input: &ReactiveInput, + _state: &dyn StateView, + ) -> Result { + Ok(HandlerOutcome { + effects: vec![ReactiveEffect::StateUpdate(StateUpdate::purge( + Address::repeat_byte(0x75), + PurgeScope::AllStorage, + ))], + quality: StateEffectQuality::RequiresRepair, + tags: vec![], + }) + } +} + +#[tokio::test] +async fn reactive_runtime_rejects_canonical_cache_effects_for_pending_inputs() -> Result<()> { + let mut cache = setup_cache().await?; + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(PendingCanonicalWriter))?; + + let err = runtime + .ingest_batch( + &mut cache, + batch(vec![( + ReactiveInput::PendingTxHash(B256::repeat_byte(0x99)), + ReactiveContext { + chain_id: Some(1), + source: InputSource::Batch, + chain_status: ChainStatus::Pending, + block: None, + transaction_index: None, + log_index: None, + }, + )]), + ) + .expect_err("pending inputs must not mutate canonical cache state"); + + assert!(matches!(err, ReactiveError::InvalidPendingEffect { .. })); + Ok(()) +} + +struct InvalidateAndResyncHandler { + address: Address, + slot: U256, +} + +impl ReactiveHandler for InvalidateAndResyncHandler { + fn id(&self) -> HandlerId { + HandlerId::new("invalidate-resync") + } + + fn interests(&self) -> Vec { + vec![ReactiveInterest::Logs(LogInterest { + provider_filter: Filter::new().address(self.address), + local_matcher: None, + route_key: Some(RouteKeySpec::EmitterAddress), + })] + } + + fn handle( + &self, + _ctx: &ReactiveContext, + _input: &ReactiveInput, + _state: &dyn StateView, + ) -> Result { + Ok(HandlerOutcome { + effects: vec![ + ReactiveEffect::Invalidate(InvalidationRequest { + scope: PurgeScope::Slots(vec![self.slot]), + address: self.address, + reason: InvalidationReason::HandlerRequested, + }), + ReactiveEffect::Resync(ResyncRequest { + id: ResyncId::new("resync-slot"), + reason: ResyncReason::HandlerRequested, + block: ResyncBlock::Hash { + number: 50, + hash: B256::repeat_byte(0x50), + require_canonical: true, + }, + targets: vec![ResyncTarget::StorageSlot { + address: self.address, + slot: self.slot, + }], + priority: ResyncPriority::High, + }), + ], + quality: StateEffectQuality::AppliedWithPendingResync, + tags: vec![], + }) + } +} + +#[tokio::test] +async fn reactive_runtime_lowers_invalidations_and_surfaces_resync_requests() -> Result<()> { + let address = Address::repeat_byte(0x76); + let slot = U256::from(11); + let mut cache = setup_cache().await?; + install_mock_erc20(&mut cache, address); + cache + .db_mut() + .insert_account_storage(address, slot, U256::from(123))?; + + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(InvalidateAndResyncHandler { address, slot }))?; + + let report = runtime.ingest_batch( + &mut cache, + batch(vec![( + ReactiveInput::Log(rpc_log(address, vec![keccak256(b"Repair()")], 50, 0, 0)), + included_context(50, 0), + )]), + )?; + + assert_eq!(report.applied.len(), 1); + assert_eq!(report.applied[0].invalidations.len(), 1); + assert_eq!(report.applied[0].resyncs.len(), 1); + assert_eq!(report.applied[0].diff.purged.len(), 1); + assert_eq!(report.resyncs.len(), 1); + assert_eq!(report.resyncs[0].priority, ResyncPriority::High); + Ok(()) +} + +struct PendingSpeculativeHandler; + +impl ReactiveHandler for PendingSpeculativeHandler { + fn id(&self) -> HandlerId { + HandlerId::new("pending-speculative") + } + + fn interests(&self) -> Vec { + vec![ReactiveInterest::PendingTransactions( + PendingTxInterest::default(), + )] + } + + fn handle( + &self, + _ctx: &ReactiveContext, + _input: &ReactiveInput, + _state: &dyn StateView, + ) -> Result { + Ok(HandlerOutcome { + effects: vec![ReactiveEffect::Speculative(SpeculativeRequest { + id: SpeculativeId::new("pending-signal"), + input_ref: InputRef::PendingTx { + chain_id: None, + hash: B256::ZERO, + }, + labels: vec![ReportTag::new("kind", "pending")], + })], + quality: StateEffectQuality::NoStateEffect, + tags: vec![], + }) + } +} + +#[tokio::test] +async fn reactive_runtime_allows_speculative_pending_effects_without_cache_mutation() -> Result<()> +{ + let mut cache = setup_cache().await?; + let mut runtime = ReactiveRuntime::::new(ReactiveConfig::default()); + runtime.register_handler(Arc::new(PendingSpeculativeHandler))?; + let tx_hash = B256::repeat_byte(0x77); + + let report = runtime.ingest_batch( + &mut cache, + batch(vec![( + ReactiveInput::PendingTxHash(tx_hash), + ReactiveContext { + chain_id: Some(1), + source: InputSource::Batch, + chain_status: ChainStatus::Pending, + block: None, + transaction_index: None, + log_index: None, + }, + )]), + )?; + + assert!(report.applied[0].state_updates.is_empty()); + assert_eq!(report.speculative.len(), 1); + assert_eq!( + report.speculative[0].input_ref, + InputRef::PendingTx { + chain_id: Some(1), + hash: tx_hash, + } + ); + Ok(()) +}