add support for operators

src/data.rs

@@ -71,6 +71,7 @@ pub struct VmNode {
     pub price: u64,
     // 1st String is user wallet and 2nd String is report message
     pub reports: HashMap<String, String>,
+    pub offline_minutes: u64,
 }
 
 impl Into<grpc::VmNodeListResp> for VmNode {
@@ -222,28 +223,74 @@ impl BrainData {
         });
     }
 
+    /// This is written to run every minute
+    pub async fn vm_nodes_cron(&self) {
+        let mut nodes = self.vm_nodes.write().unwrap();
+        let mut vm_contracts = self.vm_contracts.write().unwrap();
+        for node in nodes.iter_mut() {
+            if self.daemon_tx.contains_key(&node.public_key) {
+                node.offline_minutes = 0;
+                continue;
+            }
+            let mut operator = match self
+                .operators
+                .iter_mut()
+                .find(|o| o.vm_nodes.contains(&node.public_key))
+            {
+                Some(op) => op,
+                None => continue,
+            };
+            node.offline_minutes += 1;
+            // compensate contract admin if the node is offline more then 5 minutes
+            if node.offline_minutes > 5 {
+                for c in vm_contracts.iter() {
+                    let compensation = c.price_per_minute() * 24;
+                    if compensation < operator.escrow {
+                        operator.escrow -= compensation;
+                        self.add_nano_to_wallet(&c.admin_pubkey, compensation);
+                    }
+                }
+            }
+        }
+        // delete nodes that are offline more than 3 hours, and clean contracts
+        nodes.retain(|n| {
+            if n.offline_minutes > 180 {
+                vm_contracts.retain_mut(|c| {
+                    if c.node_pubkey == n.public_key {
+                        self.add_nano_to_wallet(&c.admin_pubkey, c.locked_nano);
+                    }
+                    c.node_pubkey != n.public_key
+                });
+                for mut op in self.operators.iter_mut() {
+                    op.vm_nodes.remove(&n.public_key);
+                }
+            }
+            n.offline_minutes <= 180
+        });
+    }
+
     pub async fn vm_contracts_cron(&self) {
         let mut deleted_contracts: Vec<(String, String)> = Vec::new();
         log::debug!("Running contracts cron...");
         {
             let mut contracts = self.vm_contracts.write().unwrap();
             contracts.retain_mut(|c| {
-                let operator_wallet = self
-                    .find_node_by_pubkey(&c.node_pubkey)
-                    .unwrap()
-                    .operator_wallet
-                    .clone();
-                let minutes_to_collect = (Utc::now() - c.collected_at).num_minutes() as u64;
-                c.collected_at = Utc::now();
-                let mut nanolp_to_collect = c.price_per_minute().saturating_mul(minutes_to_collect);
-                if nanolp_to_collect > c.locked_nano {
-                    nanolp_to_collect = c.locked_nano;
-                }
-                log::debug!("Removing {nanolp_to_collect} nanoLP from {}", c.uuid);
-                c.locked_nano -= nanolp_to_collect;
-                self.add_nano_to_wallet(&operator_wallet, nanolp_to_collect);
-                if c.locked_nano == 0 {
-                    deleted_contracts.push((c.uuid.clone(), c.node_pubkey.clone()));
+                let node = self.find_node_by_pubkey(&c.node_pubkey).unwrap();
+                if node.offline_minutes == 0 {
+                    let operator_wallet = node.operator_wallet.clone();
+                    let minutes_to_collect = (Utc::now() - c.collected_at).num_minutes() as u64;
+                    c.collected_at = Utc::now();
+                    let mut nanolp_to_collect =
+                        c.price_per_minute().saturating_mul(minutes_to_collect);
+                    if nanolp_to_collect > c.locked_nano {
+                        nanolp_to_collect = c.locked_nano;
+                    }
+                    log::debug!("Removing {nanolp_to_collect} nanoLP from {}", c.uuid);
+                    c.locked_nano -= nanolp_to_collect;
+                    self.add_nano_to_wallet(&operator_wallet, nanolp_to_collect);
+                    if c.locked_nano == 0 {
+                        deleted_contracts.push((c.uuid.clone(), c.node_pubkey.clone()));
+                    }
                 }
                 c.locked_nano > 0
             });
@@ -305,7 +352,7 @@ impl BrainData {
             minutes_to_refund = 10080;
         }
 
-        let mut refund_ammount = minutes_to_refund * contract.price_per_minute();
+        let mut refund_amount = minutes_to_refund * contract.price_per_minute();
         let mut admin_account = self
             .accounts
             .get_mut(&contract.admin_pubkey)
@@ -316,21 +363,21 @@ impl BrainData {
             .signed_duration_since(admin_account.last_kick)
             .gt(&chrono::Duration::days(1))
         {
-            refund_ammount = 0;
+            refund_amount = 0;
         }
 
-        if operator_data.escrow < refund_ammount {
-            refund_ammount = operator_data.escrow;
+        if operator_data.escrow < refund_amount {
+            refund_amount = operator_data.escrow;
         }
 
         log::debug!(
-            "Removing {refund_ammount} escrow from {} and giving it to {}",
+            "Removing {refund_amount} escrow from {} and giving it to {}",
             operator_data.key(),
             admin_account.key()
         );
-        admin_account.balance += refund_ammount;
+        admin_account.balance += refund_amount;
         admin_account.kicked_for.push(reason.to_string());
-        operator_data.escrow -= refund_ammount;
+        operator_data.escrow -= refund_amount;
 
         let admin_pubkey = contract.admin_pubkey.clone();
         drop(admin_account);
@@ -342,7 +389,7 @@ impl BrainData {
         })
         .await?;
 
-        Ok(refund_ammount)
+        Ok(refund_amount)
     }
 
     pub fn ban_user(&self, operator: &str, user: &str) {

src/main.rs

@@ -19,6 +19,7 @@ async fn main() {
     tokio::spawn(async move {
         loop {
             tokio::time::sleep(tokio::time::Duration::from_secs(60)).await;
+            data_clone.vm_nodes_cron().await;
             data_clone.vm_contracts_cron().await;
         }
     });