Use deletion batch size during delete_identities and process_identities.

src/contracts/mod.rs

@@ -146,10 +146,14 @@
         self.tree_depth
     }
 
-    pub async fn max_batch_size(&self) -> usize {
+    pub async fn max_insertion_batch_size(&self) -> usize {
         self.insertion_prover_map.read().await.max_batch_size()
     }
 
+    pub async fn max_deletion_batch_size(&self) -> usize {
+        self.deletion_prover_map.read().await.max_batch_size()
+    }
+
     #[must_use]
     pub const fn initial_leaf_value(&self) -> Field {
         self.initial_leaf_value
@@ -375,7 +379,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();
@@ -383,7 +387,7 @@
 
         tracing::error!("unpacked = {indices:?}");
 
         let padding_index = 2u32.pow(self.tree_depth as u32);
 
         Ok(indices
             .into_iter()

src/task_monitor/tasks/delete_identities.rs

@@ -61,9 +61,6 @@ async fn delete_identities(
     loop {
         let deletions = database.get_deletions().await?;
         if deletions.is_empty() {
-            // Sleep for one hour
-            // TODO: should we make this dynamic? This causes an issue with tests so its set
-            // to 1 sec atm
             tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
             continue;
         }

src/task_monitor/tasks/process_identities.rs

@@ -72,7 +72,6 @@
     identity_manager.await_clean_slate().await?;
 
     info!("Starting identity processor.");
-    let batch_size = identity_manager.max_batch_size().await;
 
     // We start a timer and force it to perform one initial tick to avoid an
     // immediate trigger.
@@ -101,6 +100,12 @@
                 // If the timer has fired we want to insert whatever
                 // identities we have, even if it's not many. This ensures
                 // a minimum quality of service for API users.
+                let batch_size = if batching_tree.peek_next_updates(1)[0].update.element == Hash::ZERO{
+                    identity_manager.max_deletion_batch_size().await
+                }else{
+                    identity_manager.max_insertion_batch_size().await
+                };
+
                 let updates = batching_tree.peek_next_updates(batch_size);
                 if updates.is_empty() {
                     continue;
@@ -138,6 +143,12 @@
                 let should_process_anyway =
                     timeout_secs.abs_diff(diff_secs) <= DEBOUNCE_THRESHOLD_SECS;
 
+                let batch_size = if batching_tree.peek_next_updates(1)[0].update.element == Hash::ZERO{
+                    identity_manager.max_deletion_batch_size().await
+                }else{
+                    identity_manager.max_insertion_batch_size().await
+                };
+
                 // We have _at most_ one complete batch here.
                 let updates = batching_tree.peek_next_updates(batch_size);
                 if updates.is_empty() {
@@ -186,37 +197,37 @@
     updates: &[AppliedTreeUpdate],
 ) -> AnyhowResult<()> {
     // If the update is an insertion
     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?;
     } 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?;
     }
 
     Ok(())
 }
@@ -255,11 +266,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;
@@ -407,28 +418,28 @@
     Ok(())
 }
 
 pub async fn delete_identities(
     database: &Database,
     identity_manager: &IdentityManager,
     batching_tree: &TreeVersion<Intermediate>,
     updates: &[AppliedTreeUpdate],
     prover: ReadOnlyProver<'_, Prover>,
 ) -> AnyhowResult<()> {
     TaskMonitor::log_identities_queues(database).await?;
 
     if updates.is_empty() {
         warn!("Identity commit requested with zero identities. Continuing.");
         return Ok(());
     }
 
     debug!("Starting identity commit for {} identities.", updates.len());
 
     // Grab the initial conditions before the updates are applied to the tree.
     let pre_root: U256 = batching_tree.get_root().into();
 
     let mut deletion_indices = updates
         .iter()
         .map(|f| f.update.leaf_index as u32)
         .collect::<Vec<u32>>();
 
     let commitments =
@@ -469,7 +480,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;
@@ -532,7 +543,7 @@
     let transaction_id = identity_manager
         .delete_identities(
             proof,
             batch_size as u32,
             packed_deletion_indices,
             pre_root,
             post_root,