Merge pull request #827 from worldcoin/piohei/fix_batch_creation

Fix batch creation.

src/task_monitor/tasks/create_batches.rs

@@ -1,11 +1,11 @@
-use std::sync::Arc;
-
 use anyhow::Context;
 use chrono::{DateTime, Utc};
 use ethers::prelude::U256;
 use ruint::Uint;
 use semaphore::merkle_tree::Proof;
 use semaphore::poseidon_tree::{Branch, PoseidonHash};
+use std::sync::Arc;
+use std::time::Duration;
 use tokio::sync::Notify;
 use tokio::{select, time};
 use tracing::instrument;
@@ -39,7 +39,7 @@ pub async fn create_batches(
 
     // We start a timer and force it to perform one initial tick to avoid an
     // immediate trigger.
-    let mut timer = time::interval(app.config.app.batch_insertion_timeout);
+    let mut timer = time::interval(Duration::from_secs(5));
     timer.tick().await;
 
     // When both futures are woken at once, the choice is made

src/task_monitor/tasks/delete_identities.rs

@@ -5,7 +5,7 @@ use std::time::Duration;
 use anyhow::Context;
 use chrono::Utc;
 use tokio::sync::{Mutex, Notify};
-use tokio::time;
+use tokio::{select, time};
 use tracing::info;
 
 use crate::app::App;
@@ -32,8 +32,11 @@ pub async fn delete_identities(
     let mut timer = time::interval(Duration::from_secs(5));
 
     loop {
-        _ = timer.tick().await;
-        info!("Deletion processor woken due to timeout");
+        select! {
+            _ = timer.tick() => {
+                info!("Deletion processor woken due to timeout");
+            }
+        }
 
         let deletions = app.database.get_deletions().await?;
         if deletions.is_empty() {

src/task_monitor/tasks/insert_identities.rs

@@ -2,7 +2,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use tokio::sync::{Mutex, Notify};
-use tokio::time;
+use tokio::{select, time};
 use tracing::info;
 
 use crate::app::App;
@@ -25,8 +25,11 @@ pub async fn insert_identities(
     let mut timer = time::interval(Duration::from_secs(5));
 
     loop {
-        _ = timer.tick().await;
-        info!("Insertion processor woken due to timeout.");
+        select! {
+            _ = timer.tick() => {
+                info!("Insertion processor woken due to timeout.");
+            }
+        }
 
         // get commits from database
         let unprocessed = app.database.get_unprocessed_commitments().await?;