From 0367d70ef34e132caa4dea6a865dc5c38693546f Mon Sep 17 00:00:00 2001 From: Noor Date: Thu, 29 May 2025 17:37:53 +0530 Subject: [PATCH] fix: refund on new_vm error refund locked_nano on daemon returns error on new_vm fix live select error on new_vm_req refactor mock daemon for return error extensive tests for new_vm --- src/db/vm.rs | 66 +++++++++++++++++++++++++++----- tests/common/vm_cli_utils.rs | 2 +- tests/common/vm_daemon_utils.rs | 10 +++-- tests/grpc_general_test.rs | 2 +- tests/grpc_vm_cli_test.rs | 67 ++++++++++++++++++++++++++++++--- tests/grpc_vm_daemon_test.rs | 2 +- 6 files changed, 127 insertions(+), 22 deletions(-) diff --git a/src/db/vm.rs b/src/db/vm.rs index f7413a1..9841aba 100644 --- a/src/db/vm.rs +++ b/src/db/vm.rs @@ -191,12 +191,50 @@ impl NewVmReq { } pub async fn submit_error(db: &Surreal, id: &str, error: String) -> Result<(), Error> { - #[derive(Serialize)] - struct NewVmError { - error: String, + let tx_query = String::from( + " + BEGIN TRANSACTION; + + LET $new_vm_req = $new_vm_req_input; + LET $error = $error_input; + LET $record = (select * from $new_vm_req)[0]; + LET $admin = $record.in; + + if $record == None {{ + THROW 'vm req not exist ' + $new_vm_req + }}; + + UPDATE $new_vm_req SET error = $error; + + UPDATE $admin SET tmp_locked -= $record.locked_nano; + UPDATE $admin SET balance += $record.locked_nano; + + COMMIT TRANSACTION;", + ); + + log::trace!("submit_new_vm_err query: {tx_query}"); + + let mut query_resp = db + .query(tx_query) + .bind(("new_vm_req_input", RecordId::from((NEW_VM_REQ, id)))) + .bind(("error_input", error)) + .await?; + + let query_err = query_resp.take_errors(); + if !query_err.is_empty() { + log::trace!("errors in submit_new_vm_err: {query_err:?}"); + let tx_fail_err_str = + String::from("The query was not executed due to a failed transaction"); + + if query_err.contains_key(&4) && query_err[&4].to_string() != tx_fail_err_str { + log::error!("vm req not exist: {}", query_err[&4]); + return Err(Error::ContractNotFound); + } else { + log::error!("Unknown error in submit_new_vm_err: {query_err:?}"); + return Err(Error::Unknown("submit_new_vm_req".to_string())); + } } - let _: Option = db.update((NEW_VM_REQ, id)).merge(NewVmError { error }).await?; - // TODO: IMP refund tmp_locked + Ok(()) } @@ -239,6 +277,9 @@ impl NewVmReq { self.public_ipv6, self.price_per_unit ); + + log::trace!("submit_new_vm_req query: {tx_query}"); + let mut query_resp = db .query(tx_query) .bind(("account_input", self.admin)) @@ -253,6 +294,7 @@ impl NewVmReq { let query_err = query_resp.take_errors(); if !query_err.is_empty() { + log::trace!("errors in submit_new_vm_req: {query_err:?}"); let tx_fail_err_str = String::from("The query was not executed due to a failed transaction"); @@ -287,14 +329,18 @@ impl WrappedMeasurement { UPDATE_VM_REQ => UPDATE_VM_REQ, _ => NEW_VM_REQ, }; - let mut resp = db - .query(format!("live select error from {table} where id = {NEW_VM_REQ}:{vm_id};")) + + let mut args_stream = db .query(format!( "live select * from measurement_args where id = measurement_args:{vm_id};" )) - .await?; - let mut error_stream = resp.stream::>(0)?; - let mut args_stream = resp.stream::>(1)?; + .await? + .stream::>(0)?; + + let mut error_stream = db + .query(format!("live select error from {table} where id = {NEW_VM_REQ}:{vm_id};")) + .await? + .stream::>(0)?; let args: Option = db.delete(("measurement_args", vm_id)).await?; diff --git a/tests/common/vm_cli_utils.rs b/tests/common/vm_cli_utils.rs index 7927410..143e86b 100644 --- a/tests/common/vm_cli_utils.rs +++ b/tests/common/vm_cli_utils.rs @@ -46,7 +46,7 @@ pub async fn create_new_vm( assert!(new_vm_resp.uuid.len() == 40); // wait for update db - tokio::time::sleep(tokio::time::Duration::from_millis(700)).await; + tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; let vm_req_db: Option = db.select((NEW_VM_REQ, new_vm_resp.uuid.clone())).await?; diff --git a/tests/common/vm_daemon_utils.rs b/tests/common/vm_daemon_utils.rs index be49c3e..dc3c367 100644 --- a/tests/common/vm_daemon_utils.rs +++ b/tests/common/vm_daemon_utils.rs @@ -8,7 +8,10 @@ use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; use tonic::transport::Channel; -pub async fn mock_vm_daemon(brain_channel: &Channel) -> Result { +pub async fn mock_vm_daemon( + brain_channel: &Channel, + daemon_error: Option, +) -> Result { let mut daemon_client = BrainVmDaemonClient::new(brain_channel.clone()); let daemon_key = Key::new(); @@ -26,7 +29,7 @@ pub async fn mock_vm_daemon(brain_channel: &Channel) -> Result { rx, )); - tokio::spawn(daemon_engine(daemon_msg_tx.clone(), brain_msg_rx)); + tokio::spawn(daemon_engine(daemon_msg_tx.clone(), brain_msg_rx, daemon_error)); tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; @@ -102,6 +105,7 @@ pub async fn daemon_msg_sender( pub async fn daemon_engine( tx: mpsc::Sender, mut rx: mpsc::Receiver, + new_vm_err: Option, ) -> Result<()> { log::info!("daemon engine vm_daemon"); while let Some(brain_msg) = rx.recv().await { @@ -120,7 +124,7 @@ pub async fn daemon_engine( let new_vm_resp = vm_proto::NewVmResp { uuid: new_vm_req.uuid.clone(), args, - error: String::new(), + error: new_vm_err.clone().unwrap_or_default(), }; let res_data = vm_proto::VmDaemonMessage { diff --git a/tests/grpc_general_test.rs b/tests/grpc_general_test.rs index 0eb9ef1..33d2b60 100644 --- a/tests/grpc_general_test.rs +++ b/tests/grpc_general_test.rs @@ -116,7 +116,7 @@ async fn test_report_node() { let db = prepare_test_db().await.unwrap(); let brain_channel = run_service_for_stream().await.unwrap(); - let daemon_key = mock_vm_daemon(&brain_channel).await.unwrap(); + let daemon_key = mock_vm_daemon(&brain_channel, None).await.unwrap(); let key = Key::new(); diff --git a/tests/grpc_vm_cli_test.rs b/tests/grpc_vm_cli_test.rs index 7338238..475e11b 100644 --- a/tests/grpc_vm_cli_test.rs +++ b/tests/grpc_vm_cli_test.rs @@ -2,24 +2,30 @@ use common::prepare_test_env::{prepare_test_db, run_service_for_stream}; use common::test_utils::Key; use common::vm_cli_utils::{airdrop, create_new_vm, user_list_vm_contracts}; use common::vm_daemon_utils::{mock_vm_daemon, register_vm_node}; +use detee_shared::vm_proto; use detee_shared::vm_proto::brain_vm_cli_client::BrainVmCliClient; use detee_shared::vm_proto::brain_vm_daemon_client::BrainVmDaemonClient; use detee_shared::vm_proto::{ExtendVmReq, ListVmContractsReq, NewVmReq}; use futures::StreamExt; use std::vec; -use surreal_brain::constants::{ACCOUNT, ACTIVE_VM}; +use surreal_brain::constants::{ACCOUNT, ACTIVE_VM, NEW_VM_REQ, TOKEN_DECIMAL}; use surreal_brain::db::prelude as db; -use surreal_brain::db::vm::ActiveVm; mod common; #[tokio::test] async fn test_vm_creation() { let db = prepare_test_db().await.unwrap(); - // env_logger::builder().filter_level(log::LevelFilter::Error).init(); + /* + env_logger::builder() + .filter_level(log::LevelFilter::Debug) + .filter_module("tungstenite", log::LevelFilter::Debug) + .filter_module("tokio_tungstenite", log::LevelFilter::Debug) + .init(); + */ let brain_channel = run_service_for_stream().await.unwrap(); - let daemon_key = mock_vm_daemon(&brain_channel).await.unwrap(); + let daemon_key = mock_vm_daemon(&brain_channel, None).await.unwrap(); let key = Key::new(); @@ -29,11 +35,60 @@ async fn test_vm_creation() { let grpc_error_message = new_vm_resp.err().unwrap().to_string(); assert!(grpc_error_message.contains("Insufficient funds")); - airdrop(&brain_channel, &key.pubkey, 10).await.unwrap(); + let airdrop_amount = 10; + airdrop(&brain_channel, &key.pubkey, airdrop_amount).await.unwrap(); let new_vm_id = create_new_vm(&db, &key, &daemon_key, &brain_channel).await.unwrap(); - let active_vm: Option = db.select((ACTIVE_VM, new_vm_id)).await.unwrap(); + let active_vm: Option = db.select((ACTIVE_VM, new_vm_id)).await.unwrap(); assert!(active_vm.is_some()); + + let daemon_key_02 = + mock_vm_daemon(&brain_channel, Some("something went wrong 04".to_string())).await.unwrap(); + + let locking_nano = 100; + let mut new_vm_req = vm_proto::NewVmReq { + admin_pubkey: key.pubkey.clone(), + node_pubkey: daemon_key_02.clone(), + price_per_unit: 1200, + extra_ports: vec![8080, 8081], + locked_nano: locking_nano, + ..Default::default() + }; + + let mut client_vm_cli = BrainVmCliClient::new(brain_channel.clone()); + let new_vm_resp = client_vm_cli + .new_vm(key.sign_request(new_vm_req.clone()).unwrap()) + .await + .unwrap() + .into_inner(); + + assert!(!new_vm_resp.error.is_empty()); + + let vm_req_db: Option = + db.select((NEW_VM_REQ, new_vm_resp.uuid.clone())).await.unwrap(); + + if let Some(new_vm_req) = vm_req_db { + assert!(!new_vm_req.error.is_empty()); + assert_eq!(new_vm_resp.error, new_vm_req.error); + } + + let acc_db: db::Account = db.select((ACCOUNT, key.pubkey.clone())).await.unwrap().unwrap(); + assert_eq!(acc_db.balance, airdrop_amount * TOKEN_DECIMAL); + assert_eq!(acc_db.tmp_locked, 0); + + new_vm_req.node_pubkey = daemon_key; + + let new_vm_resp = + client_vm_cli.new_vm(key.sign_request(new_vm_req).unwrap()).await.unwrap().into_inner(); + assert!(new_vm_resp.error.is_empty()); + + tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; + let active_vm: Option = db.select((ACTIVE_VM, new_vm_resp.uuid)).await.unwrap(); + assert!(active_vm.is_some()); + + let acc_db: db::Account = db.select((ACCOUNT, key.pubkey.clone())).await.unwrap().unwrap(); + assert_eq!(acc_db.balance, airdrop_amount * TOKEN_DECIMAL - locking_nano); + assert_eq!(acc_db.tmp_locked, 0); } #[tokio::test] diff --git a/tests/grpc_vm_daemon_test.rs b/tests/grpc_vm_daemon_test.rs index c68c5da..f8ce79a 100644 --- a/tests/grpc_vm_daemon_test.rs +++ b/tests/grpc_vm_daemon_test.rs @@ -29,7 +29,7 @@ async fn test_brain_message() { prepare_test_db().await.unwrap(); let brain_channel = run_service_for_stream().await.unwrap(); - let daemon_key = mock_vm_daemon(&brain_channel).await.unwrap(); + let daemon_key = mock_vm_daemon(&brain_channel, None).await.unwrap(); let mut cli_client = BrainVmCliClient::new(brain_channel.clone()); let cli_key = Key::new();