Merge pull request #2 from 0LNetworkCommunity/concurrency

[rust] add concurrency to v5 transaction submission

Cargo.toml

@@ -33,6 +33,7 @@ tokio = { version = "1", features = ["full"] }
 hex = "0.4.3"
 tar = "0.4.43"
 smooth-json = "0.2.7"
+futures = "0.3.31"
 
 [dev-dependencies]
 sqlx = { version = "0.8", features = [ "runtime-tokio", "tls-native-tls", "sqlite", "migrate", "macros", "derive", "postgres"] }

src/extract_transactions.rs

@@ -11,6 +11,7 @@ use libra_cached_packages::libra_stdlib::EntryFunctionCall;
 use libra_storage::read_tx_chunk::{load_chunk, load_tx_chunk_manifest};
 use libra_types::move_resource::coin_register_event::CoinRegisterEvent;
 use libra_types::util::format_signed_transaction;
+use log::{error, info, warn};
 use serde_json::json;
 use std::path::Path;
 
@@ -81,20 +82,22 @@ pub async fn extract_current_transactions(
                     make_master_tx(signed_transaction, epoch, round, timestamp, decoded_events)?;
 
                 // sanity check that we are talking about the same block, and reading vectors sequentially.
-                assert!(tx.tx_hash == tx_hash_info, "transaction hashes do not match in transaction vector and transaction_info vector");
-
-                user_txs.push(tx);
+                if tx.tx_hash != tx_hash_info {
+                    error!("transaction hashes do not match in transaction vector and transaction_info vector");
+                }
 
-                user_txs_in_chunk += 1;
+                if tx.relation_label.get_recipient().is_some() {
+                    user_txs.push(tx);
+                    user_txs_in_chunk += 1;
+                }
             }
         }
+        info!("user transactions found in chunk: {}", chunk.txns.len());
+        info!("user transactions extracted: {}", user_txs.len());
+        if user_txs_in_chunk != user_txs.len() {
+            warn!("some transactions excluded from extraction");
+        }
     }
-    // TODO: use logging
-    assert!(
-        user_txs_in_chunk == user_txs.len(),
-        "don't parse same amount of user txs as in chunk"
-    );
-    println!("user transactions found in chunk: {}", user_txs_in_chunk);
 
     Ok((user_txs, events))
 }
@@ -119,7 +122,7 @@ pub fn make_master_tx(
         }
         diem_types::transaction::TransactionPayload::Multisig(_multisig) => "Multisig".to_string(),
     };
-    let relation_label = get_relation(user_tx, &events);
+    let relation_label = pick_relation(user_tx, &events);
 
     let tx = WarehouseTxMaster {
         tx_hash,
@@ -204,20 +207,121 @@ pub fn function_args_to_json(user_tx: &SignedTransaction) -> Result<serde_json::
     Ok(json)
 }
 
-// TODO: unsure if this needs to happen on Rust side
-fn get_relation(user_tx: &SignedTransaction, events: &[WarehouseEvent]) -> RelationLabel {
-    match EntryFunctionCall::decode(user_tx.payload()) {
+fn pick_relation(user_tx: &SignedTransaction, events: &[WarehouseEvent]) -> RelationLabel {
+    if let Some(r) = maybe_get_current_relation(user_tx, events) {
+        return r;
+    }
+    if let Some(r) = maybe_get_v7_relation(user_tx, events) {
+        return r;
+    }
+
+    if let Some(r) = maybe_get_v6_relation(user_tx, events) {
+        return r;
+    }
+
+    RelationLabel::Configuration
+}
+// Using HEAD libra-framework code base try to decode transaction
+fn maybe_get_current_relation(
+    user_tx: &SignedTransaction,
+    events: &[WarehouseEvent],
+) -> Option<RelationLabel> {
+    let r = match EntryFunctionCall::decode(user_tx.payload()) {
         Some(EntryFunctionCall::OlAccountTransfer { to, amount: _ }) => {
             if is_onboarding_event(events) {
                 RelationLabel::Onboarding(to)
             } else {
                 RelationLabel::Transfer(to)
             }
         }
+        Some(EntryFunctionCall::OlAccountCreateAccount { auth_key }) => {
+            RelationLabel::Onboarding(auth_key)
+        }
+        Some(EntryFunctionCall::VouchVouchFor { friend_account }) => {
+            RelationLabel::Vouch(friend_account)
+        }
+        Some(EntryFunctionCall::VouchInsistVouchFor { friend_account }) => {
+            RelationLabel::Vouch(friend_account)
+        }
+        Some(EntryFunctionCall::CoinTransfer { to, .. }) => RelationLabel::Transfer(to),
+        Some(EntryFunctionCall::AccountRotateAuthenticationKeyWithRotationCapability {
+            rotation_cap_offerer_address,
+            ..
+        }) => RelationLabel::Transfer(rotation_cap_offerer_address),
+
         // TODO: get other entry functions with known counter parties
         // if nothing is found try to decipher from events
-        _ => RelationLabel::Tx,
-    }
+        _ => return None,
+    };
+    Some(r)
+}
+
+fn maybe_get_v7_relation(
+    user_tx: &SignedTransaction,
+    events: &[WarehouseEvent],
+) -> Option<RelationLabel> {
+    let r = match V7EntryFunctionCall::decode(user_tx.payload()) {
+        Some(V7EntryFunctionCall::OlAccountTransfer { to, amount: _ }) => {
+            if is_onboarding_event(events) {
+                RelationLabel::Onboarding(to)
+            } else {
+                RelationLabel::Transfer(to)
+            }
+        }
+        Some(V7EntryFunctionCall::OlAccountCreateAccount { auth_key }) => {
+            RelationLabel::Onboarding(auth_key)
+        }
+        Some(V7EntryFunctionCall::VouchVouchFor { friend_account }) => {
+            RelationLabel::Vouch(friend_account)
+        }
+        Some(V7EntryFunctionCall::VouchInsistVouchFor { friend_account }) => {
+            RelationLabel::Vouch(friend_account)
+        }
+        Some(V7EntryFunctionCall::CoinTransfer { to, .. }) => RelationLabel::Transfer(to),
+        Some(V7EntryFunctionCall::AccountRotateAuthenticationKeyWithRotationCapability {
+            rotation_cap_offerer_address,
+            ..
+        }) => RelationLabel::Transfer(rotation_cap_offerer_address),
+
+        // TODO: get other entry functions with known counter parties
+        // if nothing is found try to decipher from events
+        _ => return None,
+    };
+    Some(r)
+}
+
+fn maybe_get_v6_relation(
+    user_tx: &SignedTransaction,
+    events: &[WarehouseEvent],
+) -> Option<RelationLabel> {
+    let r = match V6EntryFunctionCall::decode(user_tx.payload()) {
+        Some(V6EntryFunctionCall::OlAccountTransfer { to, amount: _ }) => {
+            if is_onboarding_event(events) {
+                RelationLabel::Onboarding(to)
+            } else {
+                RelationLabel::Transfer(to)
+            }
+        }
+        Some(V6EntryFunctionCall::OlAccountCreateAccount { auth_key }) => {
+            RelationLabel::Onboarding(auth_key)
+        }
+        Some(V6EntryFunctionCall::VouchVouchFor { wanna_be_my_friend }) => {
+            RelationLabel::Vouch(wanna_be_my_friend)
+        }
+        Some(V6EntryFunctionCall::VouchInsistVouchFor { wanna_be_my_friend }) => {
+            RelationLabel::Vouch(wanna_be_my_friend)
+        }
+        Some(V6EntryFunctionCall::CoinTransfer { to, .. }) => RelationLabel::Transfer(to),
+        Some(V6EntryFunctionCall::AccountRotateAuthenticationKeyWithRotationCapability {
+            rotation_cap_offerer_address,
+            ..
+        }) => RelationLabel::Transfer(rotation_cap_offerer_address),
+
+        // TODO: get other entry functions with known counter parties
+        // if nothing is found try to decipher from events
+        _ => return None,
+    };
+    Some(r)
 }
 
 fn is_onboarding_event(events: &[WarehouseEvent]) -> bool {

src/json_rescue_v5_extract.rs

@@ -26,14 +26,15 @@ use std::path::{Path, PathBuf};
 
 pub fn extract_v5_json_rescue(
     one_json_file: &Path,
-) -> Result<(Vec<WarehouseTxMaster>, Vec<WarehouseEvent>)> {
+) -> Result<(Vec<WarehouseTxMaster>, Vec<WarehouseEvent>, Vec<String>)> {
     let json = std::fs::read_to_string(one_json_file).context("could not read file")?;
 
     let txs: Vec<TransactionViewV5> = serde_json::from_str(&json)
         .map_err(|e| anyhow!("could not parse JSON to TransactionViewV5, {:?}", e))?;
 
     let mut tx_vec = vec![];
     let event_vec = vec![];
+    let mut unique_functions = vec![];
 
     for t in txs {
         let mut wtxs = WarehouseTxMaster::default();
@@ -46,14 +47,25 @@ pub fn extract_v5_json_rescue(
 
                 wtxs.function = make_function_name(script);
                 trace!("function: {}", &wtxs.function);
+                if !unique_functions.contains(&wtxs.function) {
+                    unique_functions.push(wtxs.function.clone());
+                }
 
                 decode_transaction_args(&mut wtxs, &t.bytes)?;
 
                 // TODO:
                 // wtxs.events
                 // wtxs.block_timestamp
 
-                tx_vec.push(wtxs);
+                // TODO: create arg to exclude tx without counter party
+                match &wtxs.relation_label {
+                    RelationLabel::Tx => {}
+                    RelationLabel::Transfer(_) => tx_vec.push(wtxs),
+                    RelationLabel::Onboarding(_) => tx_vec.push(wtxs),
+                    RelationLabel::Vouch(_) => tx_vec.push(wtxs),
+                    RelationLabel::Configuration => {}
+                    RelationLabel::Miner => {}
+                };
             }
             TransactionDataView::BlockMetadata { timestamp_usecs: _ } => {
                 // TODO get epoch events
@@ -66,12 +78,18 @@ pub fn extract_v5_json_rescue(
         }
     }
 
-    Ok((tx_vec, event_vec))
+    Ok((tx_vec, event_vec, unique_functions))
 }
 
 pub fn decode_transaction_args(wtx: &mut WarehouseTxMaster, tx_bytes: &[u8]) -> Result<()> {
     // test we can bcs decode to the transaction object
-    let t: TransactionV5 = bcs::from_bytes(tx_bytes).unwrap();
+    let t: TransactionV5 = bcs::from_bytes(tx_bytes).map_err(|err| {
+        anyhow!(
+            "could not bcs decode tx_bytes, for function: {}, msg: {:?}",
+            wtx.function,
+            err
+        )
+    })?;
 
     if let TransactionV5::UserTransaction(u) = &t {
         if let TransactionPayload::ScriptFunction(_) = &u.raw_txn.payload {
@@ -84,6 +102,10 @@ pub fn decode_transaction_args(wtx: &mut WarehouseTxMaster, tx_bytes: &[u8]) ->
 
                         wtx.entry_function = Some(EntryFunctionArgs::V5(sf.to_owned()));
                     }
+                    ScriptFunctionCallGenesis::AutopayCreateInstruction { payee, .. } => {
+                        wtx.relation_label = RelationLabel::Transfer(cast_legacy_account(payee)?);
+                        wtx.entry_function = Some(EntryFunctionArgs::V5(sf.to_owned()));
+                    }
                     ScriptFunctionCallGenesis::CreateAccUser { .. } => {
                         // onboards self
                         wtx.relation_label = RelationLabel::Onboarding(wtx.sender);