Dzejkop/deletion-recovery-dedup-mining (#599)

* Split mainnet and secondary finalization

* WIP

* Fix padding issue

* Bring bck commented code

src/contracts/abi.rs

@@ -2,13 +2,41 @@
 
 use ethers::prelude::abigen;
 
+/// The `TreeChanged` event emitted by the `IdentityManager` contract.
+/// Maps to the following enum in the contract code:
+///
+/// ```sol
+/// enum TreeChange {
+///     Insertion,
+///     Deletion,
+///     Update
+/// }
+/// ```
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum TreeChangeKind {
+    Insertion,
+    Deletion,
+    Update,
+}
+
+impl From<u8> for TreeChangeKind {
+    fn from(value: u8) -> Self {
+        match value {
+            0 => Self::Insertion,
+            1 => Self::Deletion,
+            2 => Self::Update,
+            _ => panic!("Invalid value for TreeChangeKind: {}", value),
+        }
+    }
+}
+
 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 deleteIdentities(uint256[8] calldata deletionProof, uint32 batchSize, bytes calldata packedDeletionIndices, uint256 preRoot, uint256 postRoot) public virtual        
+        function deleteIdentities(uint256[8] calldata deletionProof, uint32 batchSize, bytes calldata packedDeletionIndices, uint256 preRoot, uint256 postRoot) public virtual
         function latestRoot() public view virtual returns (uint256 root)
         function owner() public view virtual returns (address)
         function queryRoot(uint256 root) public view virtual returns (RootInfo memory)

src/contracts/mod.rs

@@ -5,22 +5,23 @@ pub mod scanner;
 use std::collections::HashMap;
 use std::sync::Arc;
 
-use anyhow::anyhow;
+use anyhow::{anyhow, Context};
 use clap::Parser;
 use ethers::providers::Middleware;
-use ethers::types::{Address, U256};
+use ethers::types::{Address, H256, U256};
 use semaphore::Field;
 use tokio::sync::RwLockReadGuard;
 use tracing::{error, info, instrument, warn};
 
-use self::abi::{BridgedWorldId, WorldId};
+use self::abi::{BridgedWorldId, DeleteIdentitiesCall, WorldId};
 use crate::ethereum::write::TransactionId;
 use crate::ethereum::{Ethereum, ReadProvider};
 use crate::prover::identity::Identity;
 use crate::prover::map::{DeletionProverMap, InsertionProverMap, ReadOnlyInsertionProver};
 use crate::prover::{Proof, Prover, ProverConfiguration, ProverType, ReadOnlyProver};
 use crate::serde_utils::JsonStrWrapper;
 use crate::server::error::Error as ServerError;
+use crate::utils::index_packing::unpack_indices;
 
 /// Configuration options for the component responsible for interacting with the
 /// contract.
@@ -360,6 +361,48 @@ impl IdentityManager {
         Ok(latest_root)
     }
 
+    /// Fetches the identity commitments from a
+    /// `deleteIdentities` transaction by tx hash
+    #[instrument(level = "debug", skip_all)]
+    pub async fn fetch_deletion_indices_from_tx(
+        &self,
+        tx_hash: H256,
+    ) -> anyhow::Result<Vec<usize>> {
+        let provider = self.ethereum.provider();
+
+        let tx = provider
+            .get_transaction(tx_hash)
+            .await?
+            .context("Missing tx")?;
+
+        use ethers::abi::AbiDecode;
+        let delete_identities = DeleteIdentitiesCall::decode(&tx.input)?;
+
+        let packed_deletion_indices: &[u8] = delete_identities.packed_deletion_indices.as_ref();
+        let indices = unpack_indices(packed_deletion_indices);
+
+        tracing::error!("unpacked = {indices:?}");
+
+        let padding_index = 2u32.pow(self.tree_depth as u32);
+
+        Ok(indices
+            .into_iter()
+            .filter(|idx| *idx != padding_index)
+            .map(|x| x as usize)
+            .collect())
+    }
+
+    #[instrument(level = "debug", skip_all)]
+    pub async fn is_root_mined(&self, root: U256) -> anyhow::Result<bool> {
+        let (root_on_mainnet, ..) = self.abi.query_root(root).call().await?;
+
+        if root_on_mainnet.is_zero() {
+            return Ok(false);
+        }
+
+        Ok(true)
+    }
+
     #[instrument(level = "debug", skip_all)]
     pub async fn is_root_mined_multi_chain(&self, root: U256) -> anyhow::Result<bool> {
         let (root_on_mainnet, ..) = self.abi.query_root(root).call().await?;

src/database/mod.rs

@@ -793,7 +793,6 @@ mod test {
     use postgres_docker_utils::DockerContainerGuard;
     use ruint::Uint;
     use semaphore::Field;
-    use sqlx::Row;
 
     use super::{Database, Options};
     use crate::identity_tree::{Hash, Status};

src/identity_tree.rs

@@ -516,6 +516,23 @@ impl TreeVersion<Latest> {
     }
 }
 
+impl<T> TreeVersion<T>
+where
+    T: Version,
+{
+    pub fn commitments_by_indices(&self, indices: &[usize]) -> Vec<Hash> {
+        let tree = self.get_data();
+
+        let mut commitments = vec![];
+
+        for idx in indices {
+            commitments.push(tree.tree.get_leaf(*idx));
+        }
+
+        commitments
+    }
+}
+
 /// Public API for working with versions that have a successor. Such versions
 /// only allow peeking and applying updates from the successor.
 pub trait TreeWithNextVersion {

src/task_monitor.rs

@@ -3,7 +3,6 @@ use std::time::Duration;
 
 use anyhow::Result as AnyhowResult;
 use clap::Parser;
-use ethers::types::U256;
 use once_cell::sync::Lazy;
 use prometheus::{linear_buckets, register_gauge, register_histogram, Gauge, Histogram};
 use tokio::sync::{broadcast, Notify, RwLock};
@@ -13,19 +12,15 @@ use tracing::{info, instrument, warn};
 use self::tasks::delete_identities::DeleteIdentities;
 use self::tasks::finalize_identities::FinalizeRoots;
 use self::tasks::insert_identities::InsertIdentities;
-use self::tasks::mine_identities::MineIdentities;
 use self::tasks::process_identities::ProcessIdentities;
 use crate::contracts::SharedIdentityManager;
 use crate::database::Database;
-use crate::ethereum::write::TransactionId;
 use crate::identity_tree::TreeState;
-use crate::utils::async_queue::AsyncQueue;
 
 pub mod tasks;
 
 const PROCESS_IDENTITIES_BACKOFF: Duration = Duration::from_secs(5);
 const FINALIZE_IDENTITIES_BACKOFF: Duration = Duration::from_secs(5);
-const MINE_IDENTITIES_BACKOFF: Duration = Duration::from_secs(5);
 const INSERT_IDENTITIES_BACKOFF: Duration = Duration::from_secs(5);
 const DELETE_IDENTITIES_BACKOFF: Duration = Duration::from_secs(5);
 
@@ -34,60 +29,6 @@ struct RunningInstance {
     shutdown_sender: broadcast::Sender<()>,
 }
 
-#[derive(Debug, Clone)]
-pub struct PendingBatchInsertion {
-    transaction_id: TransactionId,
-    pre_root:       U256,
-    post_root:      U256,
-    start_index:    usize,
-}
-
-impl PendingBatchInsertion {
-    pub fn new(
-        transaction_id: TransactionId,
-        pre_root: U256,
-        post_root: U256,
-        start_index: usize,
-    ) -> Self {
-        Self {
-            transaction_id,
-            pre_root,
-            post_root,
-            start_index,
-        }
-    }
-}
-
-#[derive(Debug, Clone)]
-pub struct PendingBatchDeletion {
-    transaction_id: TransactionId,
-    pre_root:       U256,
-    commitments:    Vec<U256>,
-    post_root:      U256,
-}
-
-impl PendingBatchDeletion {
-    pub fn new(
-        transaction_id: TransactionId,
-        pre_root: U256,
-        commitments: Vec<U256>,
-        post_root: U256,
-    ) -> Self {
-        Self {
-            transaction_id,
-            pre_root,
-            commitments,
-            post_root,
-        }
-    }
-}
-
-#[derive(Debug, Clone)]
-pub enum PendingBatchSubmission {
-    Insertion(PendingBatchInsertion),
-    Deletion(PendingBatchDeletion),
-}
-
 static PENDING_IDENTITIES: Lazy<Gauge> = Lazy::new(|| {
     register_gauge!("pending_identities", "Identities not submitted on-chain").unwrap()
 });
@@ -152,11 +93,6 @@ pub struct Options {
     #[clap(long, env, default_value = "100")]
     pub insert_identities_capacity: usize,
 
-    /// How many transactions can be sent "at once" to the blockchain via the
-    /// write provider.
-    #[clap(long, env, default_value = "1")]
-    pub pending_identities_capacity: usize,
-
     /// The maximum number of windows to scan for finalization logs
     #[clap(long, env, default_value = "100")]
     pub scanning_window_size: u64,
@@ -177,12 +113,11 @@ pub struct TaskMonitor {
     /// when shutdown is called we want to be able to gracefully
     /// await the join handles - which requires ownership of the handle and by
     /// extension the instance.
-    instance:                    RwLock<Option<RunningInstance>>,
-    database:                    Arc<Database>,
-    identity_manager:            SharedIdentityManager,
-    tree_state:                  TreeState,
-    batch_insert_timeout_secs:   u64,
-    pending_identities_capacity: usize,
+    instance:                  RwLock<Option<RunningInstance>>,
+    database:                  Arc<Database>,
+    identity_manager:          SharedIdentityManager,
+    tree_state:                TreeState,
+    batch_insert_timeout_secs: u64,
 
     // Finalization params
     scanning_window_size:           u64,
@@ -202,7 +137,6 @@ impl TaskMonitor {
     ) -> Self {
         let Options {
             batch_timeout_seconds,
-            pending_identities_capacity,
             scanning_window_size,
             time_between_scans_seconds,
             batch_deletion_timeout_seconds: _,
@@ -216,7 +150,6 @@ impl TaskMonitor {
             identity_manager: contracts,
             tree_state,
             batch_insert_timeout_secs: batch_timeout_seconds,
-            pending_identities_capacity,
             scanning_window_size,
             time_between_scans: Duration::from_secs(time_between_scans_seconds),
             batch_deletion_timeout_seconds: options.batch_deletion_timeout_seconds,
@@ -240,8 +173,6 @@ impl TaskMonitor {
         // in the database
         wake_up_notify.notify_one();
 
-        let pending_batch_submissions_queue = AsyncQueue::new(self.pending_identities_capacity);
-
         let mut handles = Vec::new();
 
         // Finalize identities task
@@ -262,29 +193,12 @@ impl TaskMonitor {
 
         handles.push(finalize_identities_handle);
 
-        // Mine identities task
-        let mine_identities = MineIdentities::new(
-            self.database.clone(),
-            self.identity_manager.clone(),
-            self.tree_state.get_processed_tree(),
-            pending_batch_submissions_queue.clone(),
-        );
-
-        let mine_identities_handle = crate::utils::spawn_monitored_with_backoff(
-            move || mine_identities.clone().run(),
-            shutdown_sender.clone(),
-            MINE_IDENTITIES_BACKOFF,
-        );
-
-        handles.push(mine_identities_handle);
-
         // Process identities task
         let process_identities = ProcessIdentities::new(
             self.database.clone(),
             self.identity_manager.clone(),
             self.tree_state.get_batching_tree(),
             self.batch_insert_timeout_secs,
-            pending_batch_submissions_queue,
             wake_up_notify.clone(),
         );