Dzejkop/scan-roots (#591)

src/app.rs

@@ -475,28 +475,25 @@ impl App {
         root_state: &RootItem,
     ) -> Result<(), ServerError> {
         let latest_root = self.tree_state.get_latest_tree().get_root();
+        let batching_root = self.tree_state.get_batching_tree().get_root();
         let processed_root = self.tree_state.get_processed_tree().get_root();
         let mined_root = self.tree_state.get_mined_tree().get_root();
 
+        let root = root_state.root;
+
         match root_state.status {
-            Status::Pending if latest_root == root_state.root => return Ok(()),
+            // Pending status implies the batching or latest tree
+            Status::Pending if latest_root == root || batching_root == root => return Ok(()),
             // Processed status is hidden - this should never happen
-            Status::Processed if processed_root == root_state.root => return Ok(()),
+            Status::Processed if processed_root == root => return Ok(()),
             // Processed status is hidden so it could be either processed or mined
-            Status::Mined if processed_root == root_state.root || mined_root == root_state.root => {
-                return Ok(())
-            }
+            Status::Mined if processed_root == root || mined_root == root => return Ok(()),
             _ => (),
         }
 
-        // Latest processed root is always valid
-        if root_state.root == processed_root {
-            return Ok(());
-        }
-
         let now = chrono::Utc::now();
 
-        let root_age = if root_state.status == Status::Pending {
+        let root_age = if matches!(root_state.status, Status::Pending | Status::Processed) {
             now - root_state.pending_valid_as_of
         } else {
             let mined_at = root_state

src/contracts/abi.rs

@@ -6,6 +6,7 @@ abigen!(
     WorldId,
     r#"[
         struct RootInfo { uint256 root; uint128 supersededTimestamp; bool isValid }
+        event TreeChanged(uint256 indexed preRoot, uint8 indexed kind, uint256 indexed postRoot)
         function registerIdentities(uint256[8] calldata insertionProof, uint256 preRoot, uint32 startIndex, uint256[] calldata identityCommitments, uint256 postRoot) public virtual
         function latestRoot() public view virtual returns (uint256 root)
         function owner() public view virtual returns (address)
@@ -16,6 +17,7 @@ abigen!(
 abigen!(
     BridgedWorldId,
     r#"[
+        event RootAdded(uint256 root, uint128 timestamp)
         function rootHistory(uint256 root) public view virtual returns (uint128 timestamp)
         function latestRoot() public view returns (uint256 root)
     ]"#

src/contracts/mod.rs

@@ -1,5 +1,6 @@
 //! Functionality for interacting with smart contracts deployed on chain.
-mod abi;
+pub mod abi;
+pub mod scanner;
 
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -64,6 +65,15 @@ pub struct IdentityManager {
 }
 
 impl IdentityManager {
+    // TODO: I don't like these public getters
+    pub fn abi(&self) -> &WorldId<ReadProvider> {
+        &self.abi
+    }
+
+    pub fn secondary_abis(&self) -> &[BridgedWorldId<ReadProvider>] {
+        &self.secondary_abis
+    }
+
     #[instrument(level = "debug", skip_all)]
     pub async fn new(
         options: Options,

src/contracts/scanner.rs

@@ -0,0 +1,60 @@
+use ethers::providers::Middleware;
+use ethers::types::{Address, BlockNumber, Filter, FilterBlockOption, Log, Topic, ValueOrArray};
+
+pub struct BlockScanner<T> {
+    read_provider: T,
+    current_block: u64,
+    window_size:   u64,
+}
+
+impl<T> BlockScanner<T>
+where
+    T: Middleware,
+    <T as Middleware>::Error: 'static,
+{
+    pub async fn new_latest(read_provider: T, window_size: u64) -> anyhow::Result<Self> {
+        let latest_block = read_provider.get_block_number().await?;
+
+        Ok(Self {
+            read_provider,
+            current_block: latest_block.as_u64(),
+            window_size,
+        })
+    }
+
+    pub async fn next(
+        &mut self,
+        address: Option<ValueOrArray<Address>>,
+        topics: [Option<Topic>; 4],
+    ) -> anyhow::Result<Vec<Log>> {
+        let latest_block = self.read_provider.get_block_number().await?.as_u64();
+
+        if self.current_block >= latest_block {
+            return Ok(Vec::new());
+        }
+
+        let from_block = self.current_block;
+        let to_block = latest_block.min(from_block + self.window_size);
+
+        let next_current_block = to_block + 1;
+
+        let from_block = Some(BlockNumber::Number(from_block.into()));
+        let to_block = Some(BlockNumber::Number(to_block.into()));
+
+        let logs = self
+            .read_provider
+            .get_logs(&Filter {
+                block_option: FilterBlockOption::Range {
+                    from_block,
+                    to_block,
+                },
+                address,
+                topics,
+            })
+            .await?;
+
+        self.current_block = next_current_block;
+
+        Ok(logs)
+    }
+}

src/task_monitor.rs

@@ -94,22 +94,13 @@ pub struct Options {
     #[clap(long, env, default_value = "1")]
     pub pending_identities_capacity: usize,
 
-    /// How many roots can be held in the mined roots queue at any given time.
-    ///
-    /// There is no reason why we shouldn't be able to wait for multiple
-    /// roots to be finalized across chains at the same time.
-    ///
-    /// This is just a limit on memory usage for this channel.
-    #[clap(long, env, default_value = "10")]
-    pub mined_roots_capacity: usize,
-
-    /// The maximum number of attempts to finalize a root before giving up.
+    /// The maximum number of windows to scan for finalization logs
     #[clap(long, env, default_value = "100")]
-    pub finalization_max_attempts: usize,
+    pub scanning_window_size: u64,
 
-    /// The number of seconds to wait between attempts to finalize a root.
+    /// The number of seconds to wait between fetching logs
     #[clap(long, env, default_value = "30")]
-    pub finalization_sleep_time_seconds: u64,
+    pub time_between_scans_seconds: u64,
 }
 
 /// A worker that commits identities to the blockchain.
@@ -129,10 +120,10 @@ pub struct TaskMonitor {
     tree_state:                  TreeState,
     batch_insert_timeout_secs:   u64,
     pending_identities_capacity: usize,
-    mined_roots_capacity:        usize,
 
-    finalization_max_attempts:       usize,
-    finalization_sleep_time_seconds: u64,
+    // Finalization params
+    scanning_window_size: u64,
+    time_between_scans:   Duration,
 }
 
 impl TaskMonitor {
@@ -145,9 +136,8 @@ impl TaskMonitor {
         let Options {
             batch_timeout_seconds,
             pending_identities_capacity,
-            mined_roots_capacity,
-            finalization_max_attempts,
-            finalization_sleep_time_seconds,
+            scanning_window_size,
+            time_between_scans_seconds,
         } = *options;
 
         Self {
@@ -157,9 +147,8 @@ impl TaskMonitor {
             tree_state,
             batch_insert_timeout_secs: batch_timeout_seconds,
             pending_identities_capacity,
-            mined_roots_capacity,
-            finalization_max_attempts,
-            finalization_sleep_time_seconds,
+            scanning_window_size,
+            time_between_scans: Duration::from_secs(time_between_scans_seconds),
         }
     }
 
@@ -180,18 +169,17 @@ impl TaskMonitor {
         wake_up_notify.notify_one();
 
         let pending_batch_submissions_queue = AsyncQueue::new(self.pending_identities_capacity);
-        let mined_roots_queue = AsyncQueue::new(self.mined_roots_capacity);
 
         let mut handles = Vec::new();
 
         // Finalize identities task
         let finalize_identities = FinalizeRoots::new(
             self.database.clone(),
             self.identity_manager.clone(),
+            self.tree_state.get_processed_tree(),
             self.tree_state.get_mined_tree(),
-            mined_roots_queue.clone(),
-            self.finalization_max_attempts,
-            Duration::from_secs(self.finalization_sleep_time_seconds),
+            self.scanning_window_size,
+            self.time_between_scans,
         );
 
         let finalize_identities_handle = crate::utils::spawn_monitored_with_backoff(
@@ -208,7 +196,6 @@ impl TaskMonitor {
             self.identity_manager.clone(),
             self.tree_state.get_processed_tree(),
             pending_batch_submissions_queue.clone(),
-            mined_roots_queue,
         );
 
         let mine_identities_handle = crate::utils::spawn_monitored_with_backoff(