Intiail Commit

.gitignore

@@ -0,0 +1 @@
+/target

Cargo.toml

@@ -0,0 +1,35 @@
+[package]
+name = "grpc_banking_transactions_notifications"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+solana-sdk = "~1.16.3"
+solana-rpc-client = "~1.16.3"
+solana-rpc-client-api = "~1.16.3"
+solana-transaction-status = "~1.16.3"
+itertools = "0.10.5"
+serde = { version = "1.0.160", features = ["derive"] }
+serde_json = "1.0.96"
+bincode = "1.3.3"
+bs58 = "0.4.0"
+base64 = "0.21.0"
+thiserror = "1.0.40"
+futures = "0.3.28"
+bytes = "1.4.0"
+anyhow = "1.0.70"
+log = "0.4.17"
+clap = { version = "4.2.4", features = ["derive", "env"] }
+dashmap = "5.4.0"
+const_env = "0.1.2"
+chrono = "0.4.24"
+native-tls = "0.2.11"
+postgres-native-tls = "0.5.0"
+prometheus = "0.13.3"
+lazy_static = "1.4.0"
+
+tokio = { version = "1.32.0", features = ["rt-multi-thread", "macros", "time"] }
+yellowstone-grpc-client = { path = "../yellowstone-grpc/yellowstone-grpc-client" }
+yellowstone-grpc-proto = { path = "../yellowstone-grpc/yellowstone-grpc-proto" }

src/main.rs

@@ -0,0 +1,67 @@
+use std::{collections::HashMap, sync::Arc};
+
+use dashmap::DashMap;
+use futures::StreamExt;
+use transaction_info::TransactionInfo;
+use yellowstone_grpc_client::GeyserGrpcClient;
+use yellowstone_grpc_proto::prelude::{SubscribeRequestFilterBlocks, CommitmentLevel, subscribe_update::UpdateOneof};
+
+mod transaction_info;
+
+#[tokio::main()]
+async fn main() {
+    let grpc_addr = "http://127.0.0.1:10000";
+    let mut client = GeyserGrpcClient::connect(grpc_addr, None::<&'static str>, None).unwrap();
+    let map_of_infos = Arc::new(DashMap::<String, TransactionInfo>::new());
+
+    let mut blocks_subs = HashMap::new();
+    blocks_subs.insert(
+        "client".to_string(),
+        SubscribeRequestFilterBlocks {
+            account_include: Default::default(),
+            include_transactions: Some(true),
+            include_accounts: Some(false),
+            include_entries: Some(false),
+        },
+    );
+    let commitment_level = CommitmentLevel::Processed;
+
+    let mut stream = client
+            .subscribe_once(
+                HashMap::new(),
+                Default::default(),
+                HashMap::new(),
+                Default::default(),
+                blocks_subs,
+                Default::default(),
+                Some(commitment_level),
+                Default::default(),
+                true
+            ).await.unwrap();
+    while let Some(message) = stream.next().await {
+                let message = message.unwrap();
+    
+                let Some(update) = message.update_oneof else {
+                    continue;
+                };
+    
+                match update {
+                    UpdateOneof::BankingTransactionErrors(transaction) => {
+                        let sig = transaction.signature.to_string();
+                        match map_of_infos.get_mut(&sig) {
+                            Some(mut x) => {
+                                x.add_notification(&transaction);
+                            },
+                            None => {
+                                map_of_infos.insert(sig, TransactionInfo::new(transaction.signature, transaction.slot));
+                            }
+                        }
+                    },
+                    UpdateOneof::Block(block) => {
+                    },
+                    _ => {
+                        continue;
+                    }
+                };
+            } 
+}

src/transaction_info.rs

@@ -0,0 +1,111 @@
+use std::{collections::HashMap, hash::Hash};
+
+use solana_sdk::{slot_history::Slot, transaction::Transaction, transaction::TransactionError};
+use yellowstone_grpc_proto::prelude::SubscribeUpdateBankingTransactionResults;
+
+fn convert_transaction_error_into_int(error: &TransactionError)-> u8 {
+    match error {
+        TransactionError::AccountBorrowOutstanding=>0,
+        TransactionError::AccountInUse=>1,
+        TransactionError::AccountLoadedTwice=>2,
+        TransactionError::AccountNotFound=>3,
+        TransactionError::AddressLookupTableNotFound=>4,
+        TransactionError::AlreadyProcessed=>5,
+        TransactionError::BlockhashNotFound=>6,
+        TransactionError::CallChainTooDeep=>7,
+        TransactionError::ClusterMaintenance=>8,
+        TransactionError::DuplicateInstruction(_)=>9,
+        TransactionError::InstructionError(_, _) => 10,
+        TransactionError::InsufficientFundsForFee=>11,
+        TransactionError::InsufficientFundsForRent { .. } => 12,
+        TransactionError::InvalidAccountForFee => 13,
+        TransactionError::InvalidAccountIndex => 14,
+        TransactionError::InvalidAddressLookupTableData=>15,
+        TransactionError::InvalidAddressLookupTableIndex=>16,
+        TransactionError::InvalidAddressLookupTableOwner=>17,
+        TransactionError::InvalidLoadedAccountsDataSizeLimit=>18,
+        TransactionError::InvalidProgramForExecution=>19,
+        TransactionError::InvalidRentPayingAccount=>20,
+        TransactionError::InvalidWritableAccount=>21,
+        TransactionError::MaxLoadedAccountsDataSizeExceeded=>22,
+        TransactionError::MissingSignatureForFee=>23,
+        TransactionError::ProgramAccountNotFound=>24,
+        TransactionError::ResanitizationNeeded=>25,
+        TransactionError::SanitizeFailure=>26,
+        TransactionError::SignatureFailure=>27,
+        TransactionError::TooManyAccountLocks=>28,
+        TransactionError::UnbalancedTransaction=>29,
+        TransactionError::UnsupportedVersion=>30,
+        TransactionError::WouldExceedAccountDataBlockLimit=>31,
+        TransactionError::WouldExceedAccountDataTotalLimit=>32,
+        TransactionError::WouldExceedMaxAccountCostLimit=>33,
+        TransactionError::WouldExceedMaxBlockCostLimit=>34,
+        TransactionError::WouldExceedMaxVoteCostLimit=>35,
+    }
+}
+
+#[derive(Clone, PartialEq)]
+pub struct ErrorKey {
+    error : TransactionError,
+    slot: Slot,
+}
+
+impl Hash for ErrorKey {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        let tmp = convert_transaction_error_into_int(&self.error);
+        tmp.hash(state);
+        self.slot.hash(state);
+    }
+}
+
+impl Eq for ErrorKey {
+
+}
+
+pub struct TransactionInfo {
+    pub signature : String,
+    pub transaction_message: Option<Transaction>,
+    pub errors: HashMap<ErrorKey, usize>,
+    pub is_executed: bool,
+    pub is_confirmed : bool,
+    pub block_height : Option<u64>,
+    pub first_notification_slot: u64,
+}
+
+impl TransactionInfo {
+    pub fn new(signature: String, first_notification_slot: Slot) -> Self {
+        Self { 
+            signature, 
+            transaction_message: None,
+            errors: HashMap::new(),
+            is_executed: false,
+            is_confirmed: false,
+            first_notification_slot,
+            block_height: None,
+        }
+    }
+
+    pub fn add_notification(&mut self, notification: &SubscribeUpdateBankingTransactionResults) {
+        match &notification.error {
+            Some(error) => {
+                let slot = notification.slot;
+                let error: TransactionError = bincode::deserialize(&error.err).unwrap();
+                let key = ErrorKey {
+                    error,
+                    slot,
+                };
+                match self.errors.get_mut(&key) {
+                    Some(x) => {
+                        *x = *x + 1;
+                    },
+                    None => {
+                        self.errors.insert(key, 1);
+                    }
+                }
+            },
+            None => {
+                self.is_executed = true;
+            }
+        }
+    }
+}