udpate tests and add block ordering

Author: LeoPatOZ

tests/common/mock_callbacks.rs

@@ -0,0 +1,103 @@
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::Duration;
+
+use alloy::rpc::types::Log;
+use alloy::sol_types::SolEvent;
+use async_trait::async_trait;
+use event_scanner::EventCallback;
+use tokio::sync::Mutex;
+use tokio::time::sleep;
+
+use crate::common::TestCounter;
+
+pub struct BasicCounterCallback {
+    pub count: Arc<AtomicUsize>,
+}
+
+#[async_trait]
+impl EventCallback for BasicCounterCallback {
+    async fn on_event(&self, _log: &Log) -> anyhow::Result<()> {
+        self.count.fetch_add(1, Ordering::SeqCst);
+        Ok(())
+    }
+}
+
+pub struct SlowProcessorCallback {
+    pub delay_ms: u64,
+    pub processed: Arc<AtomicUsize>,
+}
+
+#[async_trait]
+impl EventCallback for SlowProcessorCallback {
+    async fn on_event(&self, _log: &Log) -> anyhow::Result<()> {
+        sleep(Duration::from_millis(self.delay_ms)).await;
+        self.processed.fetch_add(1, Ordering::SeqCst);
+        Ok(())
+    }
+}
+
+/// A callback that fails `fail_times` attempts before succeeding once.
+pub struct FlakyCallback {
+    pub attempts: Arc<AtomicUsize>,
+    pub successes: Arc<AtomicUsize>,
+    pub fail_times: usize,
+}
+
+#[async_trait]
+impl EventCallback for FlakyCallback {
+    async fn on_event(&self, _log: &Log) -> anyhow::Result<()> {
+        let attempt = self.attempts.fetch_add(1, Ordering::SeqCst) + 1;
+        if attempt <= self.fail_times {
+            anyhow::bail!("intentional failure on attempt {attempt}");
+        }
+        self.successes.fetch_add(1, Ordering::SeqCst);
+        Ok(())
+    }
+}
+
+// A callback that always fails and records attempts.
+pub struct AlwaysFailingCallback {
+    pub attempts: Arc<AtomicUsize>,
+}
+
+#[async_trait]
+impl EventCallback for AlwaysFailingCallback {
+    async fn on_event(&self, _log: &Log) -> anyhow::Result<()> {
+        self.attempts.fetch_add(1, Ordering::SeqCst);
+        anyhow::bail!("always failing callback")
+    }
+}
+
+// Captures block numbers in the order they are processed.
+pub struct BlockOrderingCallback {
+    pub blocks: Arc<Mutex<Vec<u64>>>,
+}
+
+#[async_trait]
+impl EventCallback for BlockOrderingCallback {
+    async fn on_event(&self, log: &Log) -> anyhow::Result<()> {
+        let mut guard = self.blocks.lock().await;
+        if let Some(n) = log.block_number {
+            guard.push(n);
+        }
+        Ok(())
+    }
+}
+
+// Captures decoded CountIncreased `newCount` values to verify callback/event ordering.
+pub struct EventOrderingCallback {
+    pub counts: Arc<Mutex<Vec<u64>>>,
+}
+
+#[async_trait]
+impl EventCallback for EventOrderingCallback {
+    async fn on_event(&self, log: &Log) -> anyhow::Result<()> {
+        if let Some(&TestCounter::CountIncreased::SIGNATURE_HASH) = log.topic0() {
+            let TestCounter::CountIncreased { newCount } = log.log_decode()?.inner.data;
+            let mut guard = self.counts.lock().await;
+            guard.push(newCount.try_into().unwrap());
+        }
+        Ok(())
+    }
+}

tests/common/mod.rs

@@ -1,19 +1,7 @@
-use std::{
-    sync::{
-        Arc,
-        atomic::{AtomicUsize, Ordering},
-    },
-    time::Duration,
-};
-
-use alloy::{
-    network::Ethereum, providers::ProviderBuilder, rpc::types::Log, sol, sol_types::SolEvent,
-};
+use alloy::{network::Ethereum, providers::ProviderBuilder, sol};
 use alloy_node_bindings::{Anvil, AnvilInstance};
-use async_trait::async_trait;
-use event_scanner::EventCallback;
-use tokio::sync::Mutex;
-use tokio::time::sleep;
+
+pub mod mock_callbacks;
 
 // Shared test contract used across integration tests
 sol! {
@@ -42,97 +30,6 @@ sol! {
     }
 }
 
-pub struct EventCounter {
-    pub count: Arc<AtomicUsize>,
-}
-
-#[async_trait]
-impl EventCallback for EventCounter {
-    async fn on_event(&self, _log: &Log) -> anyhow::Result<()> {
-        self.count.fetch_add(1, Ordering::SeqCst);
-        Ok(())
-    }
-}
-
-pub struct SlowProcessor {
-    pub delay_ms: u64,
-    pub processed: Arc<AtomicUsize>,
-}
-
-#[async_trait]
-impl EventCallback for SlowProcessor {
-    async fn on_event(&self, _log: &Log) -> anyhow::Result<()> {
-        sleep(Duration::from_millis(self.delay_ms)).await;
-        self.processed.fetch_add(1, Ordering::SeqCst);
-        Ok(())
-    }
-}
-
-/// A callback that fails `fail_times` attempts before succeeding once.
-pub struct FlakyCallback {
-    pub attempts: Arc<AtomicUsize>,
-    pub successes: Arc<AtomicUsize>,
-    pub fail_times: usize,
-}
-
-#[async_trait]
-impl EventCallback for FlakyCallback {
-    async fn on_event(&self, _log: &Log) -> anyhow::Result<()> {
-        let attempt = self.attempts.fetch_add(1, Ordering::SeqCst) + 1;
-        if attempt <= self.fail_times {
-            anyhow::bail!("intentional failure on attempt {attempt}");
-        }
-        self.successes.fetch_add(1, Ordering::SeqCst);
-        Ok(())
-    }
-}
-
-// A callback that always fails and records attempts.
-pub struct AlwaysFailingCallback {
-    pub attempts: Arc<AtomicUsize>,
-}
-
-#[async_trait]
-impl EventCallback for AlwaysFailingCallback {
-    async fn on_event(&self, _log: &Log) -> anyhow::Result<()> {
-        self.attempts.fetch_add(1, Ordering::SeqCst);
-        anyhow::bail!("always failing callback")
-    }
-}
-
-// Captures block numbers in the order they are processed.
-pub struct OrderingCapture {
-    pub blocks: Arc<Mutex<Vec<u64>>>,
-}
-
-#[async_trait]
-impl EventCallback for OrderingCapture {
-    async fn on_event(&self, log: &Log) -> anyhow::Result<()> {
-        let mut guard = self.blocks.lock().await;
-        if let Some(n) = log.block_number {
-            guard.push(n);
-        }
-        Ok(())
-    }
-}
-
-// Captures decoded CountIncreased `newCount` values to verify callback/event ordering.
-pub struct OrderingCaptureCount {
-    pub counts: Arc<Mutex<Vec<u64>>>,
-}
-
-#[async_trait]
-impl EventCallback for OrderingCaptureCount {
-    async fn on_event(&self, log: &Log) -> anyhow::Result<()> {
-        if let Some(&TestCounter::CountIncreased::SIGNATURE_HASH) = log.topic0() {
-            let TestCounter::CountIncreased { newCount } = log.log_decode()?.inner.data;
-            let mut guard = self.counts.lock().await;
-            guard.push(newCount.try_into().unwrap());
-        }
-        Ok(())
-    }
-}
-
 #[allow(clippy::missing_errors_doc)]
 pub fn spawn_anvil(block_time_secs: u64) -> anyhow::Result<AnvilInstance> {
     Ok(Anvil::new().block_time(block_time_secs).try_spawn()?)

tests/lib.rs

@@ -1,2 +1,4 @@
 pub mod common;
 pub mod live_mode;
+
+pub use common::mock_callbacks;

tests/live_mode/basic.rs

@@ -6,7 +6,10 @@ use std::{
     time::Duration,
 };
 
-use crate::common::{EventCounter, TestCounter, build_provider, deploy_counter, spawn_anvil};
+use crate::{
+    common::{TestCounter, build_provider, deploy_counter, spawn_anvil},
+    mock_callbacks::BasicCounterCallback,
+};
 use alloy::{network::Ethereum, providers::WsConnect, sol_types::SolEvent};
 use event_scanner::{event_scanner::EventScannerBuilder, types::EventFilter};
 use tokio::time::sleep;
@@ -19,7 +22,7 @@ async fn basic_single_event_scanning() -> anyhow::Result<()> {
     let contract_address = *contract.address();
 
     let event_count = Arc::new(AtomicUsize::new(0));
-    let callback = Arc::new(EventCounter { count: event_count.clone() });
+    let callback = Arc::new(BasicCounterCallback { count: event_count.clone() });
 
     let filter = EventFilter {
         contract_address,
@@ -52,8 +55,8 @@ async fn multiple_contracts_same_event_isolate_callbacks() -> anyhow::Result<()>
 
     let a_count = Arc::new(AtomicUsize::new(0));
     let b_count = Arc::new(AtomicUsize::new(0));
-    let a_cb = Arc::new(EventCounter { count: a_count.clone() });
-    let b_cb = Arc::new(EventCounter { count: b_count.clone() });
+    let a_cb = Arc::new(BasicCounterCallback { count: a_count.clone() });
+    let b_cb = Arc::new(BasicCounterCallback { count: b_count.clone() });
 
     let a_filter = EventFilter {
         contract_address: *a.address(),
@@ -96,8 +99,8 @@ async fn multiple_events_same_contract() -> anyhow::Result<()> {
 
     let increase_count = Arc::new(AtomicUsize::new(0));
     let decrease_count = Arc::new(AtomicUsize::new(0));
-    let increase_cb = Arc::new(EventCounter { count: increase_count.clone() });
-    let decrease_cb = Arc::new(EventCounter { count: decrease_count.clone() });
+    let increase_cb = Arc::new(BasicCounterCallback { count: increase_count.clone() });
+    let decrease_cb = Arc::new(BasicCounterCallback { count: decrease_count.clone() });
 
     let increase_filter = EventFilter {
         contract_address,
@@ -139,7 +142,7 @@ async fn signature_matching_ignores_irrelevant_events() -> anyhow::Result<()> {
     let contract = deploy_counter(provider).await?;
 
     let count = Arc::new(AtomicUsize::new(0));
-    let callback = Arc::new(EventCounter { count: count.clone() });
+    let callback = Arc::new(BasicCounterCallback { count: count.clone() });
 
     // Subscribe to CountDecreased but only emit CountIncreased
     let filter = EventFilter {
@@ -170,7 +173,7 @@ async fn live_filters_malformed_signature_graceful() -> anyhow::Result<()> {
     let contract = deploy_counter(provider).await?;
 
     let count = Arc::new(AtomicUsize::new(0));
-    let callback = Arc::new(EventCounter { count: count.clone() });
+    let callback = Arc::new(BasicCounterCallback { count: count.clone() });
     let filter = EventFilter {
         contract_address: *contract.address(),
         event: "invalid-sig".to_string(),

tests/live_mode/callbacks.rs

@@ -1,8 +1,20 @@
-use std::{sync::{Arc, atomic::{AtomicUsize, Ordering}}, time::Duration};
-
+use std::{
+    sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    },
+    time::Duration,
+};
+
+use crate::{
+    common::{TestCounter, build_provider, deploy_counter, spawn_anvil},
+    mock_callbacks::{AlwaysFailingCallback, FlakyCallback, SlowProcessorCallback},
+};
 use alloy::{network::Ethereum, providers::WsConnect, sol_types::SolEvent};
-use crate::common::{TestCounter, build_provider, deploy_counter, spawn_anvil, SlowProcessor, FlakyCallback, AlwaysFailingCallback};
-use event_scanner::{types::{EventFilter, CallbackConfig}, event_scanner::EventScannerBuilder};
+use event_scanner::{
+    event_scanner::EventScannerBuilder,
+    types::{CallbackConfig, EventFilter},
+};
 use tokio::time::sleep;
 
 #[tokio::test]
@@ -13,9 +25,13 @@ async fn callbacks_slow_processing_does_not_drop_events() -> anyhow::Result<()>
     let contract_address = *contract.address();
 
     let processed = Arc::new(AtomicUsize::new(0));
-    let callback = Arc::new(SlowProcessor { delay_ms: 100, processed: processed.clone() });
+    let callback = Arc::new(SlowProcessorCallback { delay_ms: 100, processed: processed.clone() });
 
-    let filter = EventFilter { contract_address, event: TestCounter::CountIncreased::SIGNATURE.to_owned(), callback };
+    let filter = EventFilter {
+        contract_address,
+        event: TestCounter::CountIncreased::SIGNATURE.to_owned(),
+        callback,
+    };
     let builder = EventScannerBuilder::<Ethereum>::new().with_event_filter(filter);
     let mut scanner = builder.connect_ws(WsConnect::new(anvil.ws_endpoint_url())).await?;
 
@@ -42,12 +58,21 @@ async fn callbacks_failure_then_retry_success() -> anyhow::Result<()> {
 
     let attempts = Arc::new(AtomicUsize::new(0));
     let successes = Arc::new(AtomicUsize::new(0));
-    let callback = Arc::new(FlakyCallback { attempts: attempts.clone(), successes: successes.clone(), fail_times: 2 });
-
-    let filter = EventFilter { contract_address: *contract.address(), event: TestCounter::CountIncreased::SIGNATURE.to_owned(), callback };
+    let callback = Arc::new(FlakyCallback {
+        attempts: attempts.clone(),
+        successes: successes.clone(),
+        fail_times: 2,
+    });
+
+    let filter = EventFilter {
+        contract_address: *contract.address(),
+        event: TestCounter::CountIncreased::SIGNATURE.to_owned(),
+        callback,
+    };
     let cfg = CallbackConfig { max_attempts: 3, delay_ms: 50 };
 
-    let builder = EventScannerBuilder::<Ethereum>::new().with_event_filter(filter).with_callback_config(cfg);
+    let builder =
+        EventScannerBuilder::<Ethereum>::new().with_event_filter(filter).with_callback_config(cfg);
     let mut scanner = builder.connect_ws(WsConnect::new(anvil.ws_endpoint_url())).await?;
     let scanner_handle = tokio::spawn(async move { scanner.start().await });
 
@@ -69,9 +94,14 @@ async fn callbacks_always_failing_respects_max_attempts() -> anyhow::Result<()>
     let attempts = Arc::new(AtomicUsize::new(0));
     let callback = Arc::new(AlwaysFailingCallback { attempts: attempts.clone() });
 
-    let filter = EventFilter { contract_address: *contract.address(), event: TestCounter::CountIncreased::SIGNATURE.to_owned(), callback };
+    let filter = EventFilter {
+        contract_address: *contract.address(),
+        event: TestCounter::CountIncreased::SIGNATURE.to_owned(),
+        callback,
+    };
     let cfg = CallbackConfig { max_attempts: 2, delay_ms: 20 };
-    let builder = EventScannerBuilder::<Ethereum>::new().with_event_filter(filter).with_callback_config(cfg);
+    let builder =
+        EventScannerBuilder::<Ethereum>::new().with_event_filter(filter).with_callback_config(cfg);
     let mut scanner = builder.connect_ws(WsConnect::new(anvil.ws_endpoint_url())).await?;
     let scanner_handle = tokio::spawn(async move { scanner.start().await });