upgrade processor to grpc 1.4 (node 1.10) (#285)

* upgrade processor to 1.10

* remove restriction on signature field

* lint

rust/Cargo.toml

@@ -26,7 +26,7 @@ aptos-moving-average = { path = "moving-average" }
 ahash = { version = "0.8.7", features = ["serde"] }
 anyhow = "1.0.62"
 async-trait = "0.1.53"
-aptos-protos = { git = "https://github.com/aptos-labs/aptos-core.git", rev = "af0dcea7144225a709e4f595e58f8026b99e901c" }
+aptos-protos = { git = "https://github.com/aptos-labs/aptos-core.git", rev = "5d1cefad0ea0d37fb08e9aec7847080745c83f9c" }
 backtrace = "0.3.58"
 base64 = "0.13.0"
 bb8 = "0.8.1"
@@ -45,7 +45,7 @@ diesel_migrations = { version = "2.1.0", features = ["postgres"] }
 diesel_async_migrations = { git = "https://github.com/niroco/diesel_async_migrations", rev = "11f331b73c5cfcc894380074f748d8fda710ac12" }
 enum_dispatch = "0.3.12"
 field_count = "0.1.1"
-futures = "0.3.24" # Previously futures v0.3.23 caused some consensus network_tests to fail. We now pin the dependency to v0.3.24.
+futures = "0.3.30"
 futures-core = "0.3.25"
 futures-util = "0.3.21"
 gcloud-sdk = { version = "0.20.4", features = [

rust/processor/migrations/2024-02-16-234847_any_signature/down.sql

@@ -0,0 +1,3 @@
+-- This file should undo anything in `up.sql`
+ALTER TABLE signatures
+ALTER COLUMN signature TYPE VARCHAR(200);

rust/processor/migrations/2024-02-16-234847_any_signature/up.sql

@@ -0,0 +1,3 @@
+-- Your SQL goes here
+ALTER TABLE signatures
+ALTER COLUMN signature TYPE text;

rust/processor/src/models/default_models/transactions.rs

@@ -273,7 +273,22 @@ impl Transaction {
                     wsc_detail,
                 )
             },
-            TxnData::StateCheckpoint(_state_checkpoint_txn) => (
+            TxnData::StateCheckpoint(_) => (
+                Self::from_transaction_info_with_data(
+                    transaction_info,
+                    None,
+                    None,
+                    version,
+                    transaction_type,
+                    0,
+                    block_height,
+                    epoch,
+                ),
+                None,
+                vec![],
+                vec![],
+            ),
+            TxnData::Validator(_) => (
                 Self::from_transaction_info_with_data(
                     transaction_info,
                     None,

rust/processor/src/models/user_transactions_models/signatures.rs

@@ -5,14 +5,14 @@
 
 use crate::{
     schema::signatures::{self},
-    utils::{counters::PROCESSOR_DATA_GAP_COUNT, util::standardize_address},
+    utils::{counters::PROCESSOR_UNKNOWN_TYPE_COUNT, util::standardize_address},
 };
 use anyhow::{Context, Result};
 use aptos_protos::transaction::v1::{
-    account_signature::Signature as AccountSignatureEnum,
-    any_signature::Type as AnySignatureTypeEnumPb, signature::Signature as SignatureEnum,
-    AccountSignature as ProtoAccountSignature, Ed25519Signature as Ed25519SignaturePB,
-    FeePayerSignature as ProtoFeePayerSignature, MultiAgentSignature as ProtoMultiAgentSignature,
+    account_signature::Signature as AccountSignatureEnum, any_signature::SignatureVariant,
+    signature::Signature as SignatureEnum, AccountSignature as ProtoAccountSignature,
+    Ed25519Signature as Ed25519SignaturePB, FeePayerSignature as ProtoFeePayerSignature,
+    MultiAgentSignature as ProtoMultiAgentSignature,
     MultiEd25519Signature as MultiEd25519SignaturePb, MultiKeySignature as MultiKeySignaturePb,
     Signature as TransactionSignaturePb, SingleKeySignature as SingleKeySignaturePb,
     SingleSender as SingleSenderPb,
@@ -337,22 +337,10 @@ impl Signature {
     ) -> Self {
         let signer = standardize_address(override_address.unwrap_or(sender));
         let signature = s.signature.as_ref().unwrap();
-        let type_ = match AnySignatureTypeEnumPb::try_from(signature.r#type) {
-            Ok(AnySignatureTypeEnumPb::Ed25519) => String::from("single_key_ed25519_signature"),
-            Ok(AnySignatureTypeEnumPb::Secp256k1Ecdsa) => {
-                String::from("single_key_secp256k1_ecdsa_signature")
-            },
-            wildcard => {
-                tracing::warn!(
-                    "Unspecified signature type or un-recognized type is not supported: {:?}",
-                    wildcard
-                );
-                PROCESSOR_DATA_GAP_COUNT
-                    .with_label_values(&["unspecified_signature_type"])
-                    .inc();
-                "".to_string()
-            },
-        };
+        let signature_bytes =
+            Self::get_any_signature_bytes(&signature.signature_variant, transaction_version);
+        let type_ =
+            Self::get_any_signature_type(&signature.signature_variant, true, transaction_version);
         Self {
             transaction_version,
             transaction_block_height,
@@ -365,7 +353,7 @@ impl Signature {
             ),
             threshold: 1,
             public_key_indices: serde_json::Value::Array(vec![]),
-            signature: format!("0x{}", hex::encode(signature.signature.as_slice())),
+            signature: format!("0x{}", hex::encode(signature_bytes.as_slice())),
             multi_agent_index,
             multi_sig_index: 0,
         }
@@ -394,25 +382,15 @@ impl Signature {
                 .unwrap()
                 .public_key
                 .clone();
-            let signature_bytes = signature.signature.as_ref().unwrap().signature.clone();
-            let type_ = match AnySignatureTypeEnumPb::try_from(
-                signature.signature.as_ref().unwrap().r#type,
-            ) {
-                Ok(AnySignatureTypeEnumPb::Ed25519) => String::from("multi_key_ed25519_signature"),
-                Ok(AnySignatureTypeEnumPb::Secp256k1Ecdsa) => {
-                    String::from("multi_key_secp256k1_ecdsa_signature")
-                },
-                wildcard => {
-                    tracing::warn!(
-                        "Unspecified signature type or un-recognized type is not supported: {:?}",
-                        wildcard
-                    );
-                    PROCESSOR_DATA_GAP_COUNT
-                        .with_label_values(&["unspecified_signature_type"])
-                        .inc();
-                    "".to_string()
-                },
-            };
+            let signature_bytes = Self::get_any_signature_bytes(
+                &signature.signature.as_ref().unwrap().signature_variant,
+                transaction_version,
+            );
+            let type_ = Self::get_any_signature_type(
+                &signature.signature.as_ref().unwrap().signature_variant,
+                false,
+                transaction_version,
+            );
             signatures.push(Self {
                 transaction_version,
                 transaction_block_height,
@@ -437,6 +415,58 @@ impl Signature {
         signatures
     }
 
+    fn get_any_signature_bytes(
+        signature_variant: &Option<SignatureVariant>,
+        transaction_version: i64,
+    ) -> Vec<u8> {
+        match signature_variant {
+            Some(SignatureVariant::Ed25519(sig)) => sig.signature.clone(),
+            Some(SignatureVariant::Zkid(sig)) => sig.signature.clone(),
+            Some(SignatureVariant::Webauthn(sig)) => sig.signature.clone(),
+            Some(SignatureVariant::Secp256k1Ecdsa(sig)) => sig.signature.clone(),
+            None => {
+                PROCESSOR_UNKNOWN_TYPE_COUNT
+                    .with_label_values(&["SignatureVariant"])
+                    .inc();
+                tracing::warn!(
+                    transaction_version = transaction_version,
+                    "Signature variant doesn't exist",
+                );
+                0u8.to_be_bytes().to_vec()
+            },
+        }
+    }
+
+    fn get_any_signature_type(
+        signature_variant: &Option<SignatureVariant>,
+        is_single_sender: bool,
+        transaction_version: i64,
+    ) -> String {
+        let prefix = if is_single_sender {
+            "single_sender"
+        } else {
+            "multi_key"
+        };
+        match signature_variant {
+            Some(SignatureVariant::Ed25519(_)) => format!("{}_ed25519_signature", prefix),
+            Some(SignatureVariant::Zkid(_)) => format!("{}_zkid_signature", prefix),
+            Some(SignatureVariant::Webauthn(_)) => format!("{}_webauthn_signature", prefix),
+            Some(SignatureVariant::Secp256k1Ecdsa(_)) => {
+                format!("{}_secp256k1_ecdsa_signature", prefix)
+            },
+            None => {
+                PROCESSOR_UNKNOWN_TYPE_COUNT
+                    .with_label_values(&["SignatureVariant"])
+                    .inc();
+                tracing::warn!(
+                    transaction_version = transaction_version,
+                    "Signature variant doesn't exist",
+                );
+                "unknown".to_string()
+            },
+        }
+    }
+
     fn parse_single_sender(
         s: &SingleSenderPb,
         sender: &String,

rust/processor/src/schema.rs

@@ -906,8 +906,7 @@ diesel::table! {
         type_ -> Varchar,
         #[max_length = 136]
         public_key -> Varchar,
-        #[max_length = 200]
-        signature -> Varchar,
+        signature -> Text,
         threshold -> Int8,
         public_key_indices -> Jsonb,
         inserted_at -> Timestamp,

rust/processor/src/utils/util.rs

@@ -1,7 +1,10 @@
 // Copyright © Aptos Foundation
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::models::property_map::{PropertyMap, TokenObjectPropertyMap};
+use crate::{
+    models::property_map::{PropertyMap, TokenObjectPropertyMap},
+    utils::counters::PROCESSOR_UNKNOWN_TYPE_COUNT,
+};
 use aptos_protos::{
     transaction::v1::{
         multisig_transaction_payload::Payload as MultisigPayloadType,
@@ -114,6 +117,16 @@ pub fn get_payload_type(payload: &TransactionPayload) -> String {
 /// This function converts the string into json recursively and lets the diesel ORM handles
 /// the escaping.
 pub fn get_clean_payload(payload: &TransactionPayload, version: i64) -> Option<Value> {
+    if payload.payload.as_ref().is_none() {
+        PROCESSOR_UNKNOWN_TYPE_COUNT
+            .with_label_values(&["TransactionPayload"])
+            .inc();
+        tracing::warn!(
+            transaction_version = version,
+            "Transaction payload doesn't exist",
+        );
+        return None;
+    }
     match payload.payload.as_ref().unwrap() {
         PayloadType::EntryFunctionPayload(inner) => {
             let clean = get_clean_entry_function_payload(inner, version);
@@ -129,12 +142,6 @@ pub fn get_clean_payload(payload: &TransactionPayload, version: i64) -> Option<V
                 panic!()
             }))
         },
-        PayloadType::ModuleBundlePayload(inner) => {
-            Some(serde_json::to_value(inner).unwrap_or_else(|_| {
-                tracing::error!(version = version, "Unable to serialize payload into value");
-                panic!()
-            }))
-        },
         PayloadType::WriteSetPayload(inner) => {
             if let Some(writeset) = inner.write_set.as_ref() {
                 get_clean_writeset(writeset, version)