Bring back tx monitoring

src/contracts/mod.rs

@@ -323,7 +323,7 @@
     }
 
     #[instrument(level = "debug", skip(self))]
-    pub async fn mine_identities(&self, transaction_id: TransactionId) -> anyhow::Result<bool> {
+    pub async fn mine_transaction(&self, transaction_id: TransactionId) -> anyhow::Result<bool> {
         let result = self.ethereum.mine_transaction(transaction_id).await?;
 
         Ok(result)
@@ -345,7 +345,7 @@
         for pending_identity_tx in pending_identities {
             // Ignores the result of each transaction - we only care about a clean slate in
             // terms of pending transactions
-            drop(self.mine_identities(pending_identity_tx).await);
+            drop(self.mine_transaction(pending_identity_tx).await);
         }
 
         Ok(())
@@ -372,7 +372,7 @@
             .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();
@@ -380,7 +380,7 @@
 
         tracing::error!("unpacked = {indices:?}");
 
         let padding_index = 2u32.pow(self.tree_depth as u32);
 
         Ok(indices
             .into_iter()

src/task_monitor.rs

@@ -5,13 +5,14 @@ use anyhow::Result as AnyhowResult;
 use clap::Parser;
 use once_cell::sync::Lazy;
 use prometheus::{linear_buckets, register_gauge, register_histogram, Gauge, Histogram};
-use tokio::sync::{broadcast, Notify, RwLock};
+use tokio::sync::{broadcast, mpsc, Notify, RwLock};
 use tokio::task::JoinHandle;
 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::monitor_txs::MonitorTxs;
 use self::tasks::process_identities::ProcessIdentities;
 use crate::contracts::SharedIdentityManager;
 use crate::database::Database;
@@ -98,19 +99,17 @@ pub struct Options {
     #[clap(long, env, default_value = "0")]
     pub max_epoch_duration_seconds: u64,
 
-    /// How many identities can be held in the API insertion queue at any given
-    /// time Past this limit the API request will block until the queue has
-    /// space for the insertion.
-    #[clap(long, env, default_value = "100")]
-    pub insert_identities_capacity: usize,
-
     /// The maximum number of windows to scan for finalization logs
     #[clap(long, env, default_value = "100")]
     pub scanning_window_size: u64,
 
     /// The number of seconds to wait between fetching logs
     #[clap(long, env, default_value = "30")]
     pub time_between_scans_seconds: u64,
+
+    /// The number of txs in the channel that we'll be monitoring
+    #[clap(long, env, default_value = "100")]
+    pub monitored_txs_capacity: usize,
 }
 
 /// A worker that commits identities to the blockchain.
@@ -138,6 +137,7 @@ pub struct TaskMonitor {
     batch_deletion_timeout_seconds: i64,
     // TODO: docs
     min_batch_deletion_size:        usize,
+    monitored_txs_capacity:         usize,
 }
 
 impl TaskMonitor {
@@ -151,10 +151,10 @@ impl TaskMonitor {
             batch_timeout_seconds,
             scanning_window_size,
             time_between_scans_seconds,
-            batch_deletion_timeout_seconds: _,
-            min_batch_deletion_size: _,
-            insert_identities_capacity: _,
             max_epoch_duration_seconds,
+            monitored_txs_capacity,
+            batch_deletion_timeout_seconds,
+            min_batch_deletion_size,
         } = *options;
 
         Self {
@@ -165,9 +165,10 @@ impl TaskMonitor {
             batch_insert_timeout_secs: batch_timeout_seconds,
             scanning_window_size,
             time_between_scans: Duration::from_secs(time_between_scans_seconds),
-            batch_deletion_timeout_seconds: options.batch_deletion_timeout_seconds,
-            min_batch_deletion_size: options.min_batch_deletion_size,
+            batch_deletion_timeout_seconds,
+            min_batch_deletion_size,
             max_epoch_duration: Duration::from_secs(max_epoch_duration_seconds),
+            monitored_txs_capacity,
         }
     }
 
@@ -182,6 +183,9 @@ impl TaskMonitor {
         // but for symmetry's sake we create it for every task with `.subscribe()`
         let (shutdown_sender, _) = broadcast::channel(1);
 
+        let (monitored_txs_sender, monitored_txs_receiver) =
+            mpsc::channel(self.monitored_txs_capacity);
+
         let wake_up_notify = Arc::new(Notify::new());
         // Immediately notify so we can start processing if we have pending identities
         // in the database
@@ -214,6 +218,7 @@ impl TaskMonitor {
             self.identity_manager.clone(),
             self.tree_state.get_batching_tree(),
             self.batch_insert_timeout_secs,
+            monitored_txs_sender,
             wake_up_notify.clone(),
         );
 
@@ -225,6 +230,16 @@ impl TaskMonitor {
 
         handles.push(process_identities_handle);
 
+        let monitor_txs = MonitorTxs::new(self.identity_manager.clone(), monitored_txs_receiver);
+
+        let monitor_txs_handle = crate::utils::spawn_monitored_with_backoff(
+            move || monitor_txs.clone().run(),
+            shutdown_sender.clone(),
+            PROCESS_IDENTITIES_BACKOFF,
+        );
+
+        handles.push(monitor_txs_handle);
+
         // Insert identities task
         let insert_identities = InsertIdentities::new(
             self.database.clone(),

src/task_monitor/tasks/mod.rs

@@ -1,4 +1,5 @@
 pub mod delete_identities;
 pub mod finalize_identities;
 pub mod insert_identities;
+pub mod monitor_txs;
 pub mod process_identities;

src/task_monitor/tasks/monitor_txs.rs

@@ -0,0 +1,45 @@
+use std::sync::Arc;
+
+use anyhow::Result as AnyhowResult;
+use tokio::sync::{mpsc, Mutex};
+
+use crate::contracts::{IdentityManager, SharedIdentityManager};
+use crate::ethereum::write::TransactionId;
+
+pub struct MonitorTxs {
+    identity_manager:       SharedIdentityManager,
+    monitored_txs_receiver: Arc<Mutex<mpsc::Receiver<TransactionId>>>,
+}
+
+impl MonitorTxs {
+    pub fn new(
+        identity_manager: SharedIdentityManager,
+        monitored_txs_receiver: mpsc::Receiver<TransactionId>,
+    ) -> Arc<Self> {
+        Arc::new(Self {
+            identity_manager,
+            monitored_txs_receiver: Arc::new(Mutex::new(monitored_txs_receiver)),
+        })
+    }
+
+    pub async fn run(self: Arc<Self>) -> anyhow::Result<()> {
+        monitor_txs_loop(&self.identity_manager, &self.monitored_txs_receiver).await?;
+
+        Ok(())
+    }
+}
+
+async fn monitor_txs_loop(
+    identity_manager: &IdentityManager,
+    monitored_txs_receiver: &Mutex<mpsc::Receiver<TransactionId>>,
+) -> AnyhowResult<()> {
+    let mut monitored_txs_receiver = monitored_txs_receiver.lock().await;
+
+    while let Some(tx) = monitored_txs_receiver.recv().await {
+        if !identity_manager.mine_transaction(tx.clone()).await? {
+            panic!("Failed to mine transaction: {}", tx);
+        }
+    }
+
+    Ok(())
+}

src/task_monitor/tasks/process_identities.rs

@@ -7,12 +7,13 @@
 use ruint::Uint;
 use semaphore::merkle_tree::Proof;
 use semaphore::poseidon_tree::Branch;
-use tokio::sync::Notify;
+use tokio::sync::{mpsc, Notify};
 use tokio::{select, time};
 use tracing::{debug, error, info, instrument, warn};
 
 use crate::contracts::{IdentityManager, SharedIdentityManager};
 use crate::database::Database;
+use crate::ethereum::write::TransactionId;
 use crate::identity_tree::{
     AppliedTreeUpdate, Hash, Intermediate, TreeVersion, TreeVersionReadOps, TreeWithNextVersion,
 };
@@ -30,6 +31,7 @@
     identity_manager:          SharedIdentityManager,
     batching_tree:             TreeVersion<Intermediate>,
     batch_insert_timeout_secs: u64,
+    monitored_txs_sender:      mpsc::Sender<TransactionId>,
     wake_up_notify:            Arc<Notify>,
 }
 
@@ -39,13 +41,15 @@
         identity_manager: SharedIdentityManager,
         batching_tree: TreeVersion<Intermediate>,
         batch_insert_timeout_secs: u64,
+        monitored_txs_sender: mpsc::Sender<TransactionId>,
         wake_up_notify: Arc<Notify>,
     ) -> Arc<Self> {
         Arc::new(Self {
             database,
             identity_manager,
             batching_tree,
             batch_insert_timeout_secs,
+            monitored_txs_sender,
             wake_up_notify,
         })
     }
@@ -55,6 +59,7 @@
             &self.database,
             &self.identity_manager,
             &self.batching_tree,
+            &self.monitored_txs_sender,
             &self.wake_up_notify,
             self.batch_insert_timeout_secs,
         )
@@ -66,6 +71,7 @@
     database: &Database,
     identity_manager: &IdentityManager,
     batching_tree: &TreeVersion<Intermediate>,
+    monitored_txs_sender: &mpsc::Sender<TransactionId>,
     wake_up_notify: &Notify,
     timeout_secs: u64,
 ) -> AnyhowResult<()> {
@@ -121,6 +127,7 @@
                     database,
                     identity_manager,
                     batching_tree,
+                    monitored_txs_sender,
                     &updates,
                 ).await?;
 
@@ -180,6 +187,7 @@
                     database,
                     identity_manager,
                     batching_tree,
+                    monitored_txs_sender,
                     &updates,
                 ).await?;
 
@@ -201,39 +209,44 @@
     database: &Database,
     identity_manager: &IdentityManager,
     batching_tree: &TreeVersion<Intermediate>,
+    monitored_txs_sender: &mpsc::Sender<TransactionId>,
     updates: &[AppliedTreeUpdate],
 ) -> AnyhowResult<()> {
     // If the update is an insertion
-    if updates
+    let tx_id = if updates
         .first()
         .context("Updates should be > 1")?
         .update
         .element
         != Hash::ZERO
     {
         let prover = identity_manager
             .get_suitable_insertion_prover(updates.len())
             .await?;
 
         info!(
             "Sending timed-out insertion batch with {}/{} updates.",
             updates.len(),
             prover.batch_size()
         );
 
-        insert_identities(database, identity_manager, batching_tree, updates, prover).await?;
+        insert_identities(database, identity_manager, batching_tree, updates, prover).await?
     } else {
         let prover = identity_manager
             .get_suitable_deletion_prover(updates.len())
             .await?;
 
         info!(
             "Sending timed-out deletion batch with {}/{} updates.",
             updates.len(),
             prover.batch_size()
         );
 
-        delete_identities(database, identity_manager, batching_tree, updates, prover).await?;
+        delete_identities(database, identity_manager, batching_tree, updates, prover).await?
+    };
+
+    if let Some(tx_id) = tx_id {
+        monitored_txs_sender.send(tx_id).await?;
     }
 
     Ok(())
@@ -246,12 +259,12 @@
     batching_tree: &TreeVersion<Intermediate>,
     updates: &[AppliedTreeUpdate],
     prover: ReadOnlyProver<'_, Prover>,
-) -> AnyhowResult<()> {
+) -> AnyhowResult<Option<TransactionId>> {
     TaskMonitor::log_identities_queues(database).await?;
 
     if updates.is_empty() {
         warn!("Identity commit requested with zero identities. Continuing.");
-        return Ok(());
+        return Ok(None);
     }
 
     debug!("Starting identity commit for {} identities.", updates.len());
@@ -273,11 +286,11 @@
                 .map(|update| update.update.element)
                 .collect::<Vec<_>>();
 
             panic!(
                 "Identities are not consecutive leaves in the tree (leaf_indexes = {:?}, \
                  commitments = {:?})",
                 leaf_indexes, commitments
             );
         }
 
         last_index = update.update.leaf_index;
@@ -422,7 +435,7 @@
 
     TaskMonitor::log_batch_size(updates.len());
 
-    Ok(())
+    Ok(Some(transaction_id))
 }
 
 pub async fn delete_identities(
@@ -431,12 +444,12 @@
     batching_tree: &TreeVersion<Intermediate>,
     updates: &[AppliedTreeUpdate],
     prover: ReadOnlyProver<'_, Prover>,
-) -> AnyhowResult<()> {
+) -> AnyhowResult<Option<TransactionId>> {
     TaskMonitor::log_identities_queues(database).await?;
 
     if updates.is_empty() {
         warn!("Identity commit requested with zero identities. Continuing.");
-        return Ok(());
+        return Ok(None);
     }
 
     debug!("Starting identity commit for {} identities.", updates.len());
@@ -446,7 +459,7 @@
 
     let mut deletion_indices = updates
         .iter()
         .map(|f| f.update.leaf_index as u32)
         .collect::<Vec<u32>>();
 
     let commitments =
@@ -487,7 +500,7 @@
     // ensure that our batches match that size. We do this by padding deletion
     // indices with tree.depth() ^ 2. The deletion prover will skip the proof for
     // any deletion with an index greater than the max tree depth
     let pad_index = 2_u32.pow(latest_tree_from_updates.depth() as u32);
 
     if commitment_count != batch_size {
         let padding = batch_size - commitment_count;
@@ -569,5 +582,5 @@
 
     TaskMonitor::log_batch_size(updates.len());
 
-    Ok(())
+    Ok(Some(transaction_id))
 }