feat: get_txs_by_signatures implementation

.github/workflows/rust.yml

@@ -19,7 +19,7 @@ jobs:
           ~/.cargo/registry/
           ~/.cargo/git/
           target/
-        key: ${{ runner.os }}-cargo-utility-chain-1.75
+        key: ${{ runner.os }}-cargo-utility-chain-1.75v2
 
     - name: Install dependencies
       run: sudo apt-get update && sudo apt-get install -y protobuf-compiler

backfill_rpc/Cargo.toml

@@ -14,5 +14,11 @@ entities = { path = "../entities" }
 solana-sdk = "~1.16"
 solana-client = "~1.16"
 solana-program = "~1.16"
+solana-transaction-status = "~1.16"
 interface = { path = "../interface" }
 tokio = { version = "1.26.0", features = ["full"] }
+plerkle_serialization = "1.5.2"
+flatbuffers = "23.1.21"
+
+[features]
+rpc_tests = []

backfill_rpc/src/rpc.rs

@@ -1,28 +1,34 @@
 use async_trait::async_trait;
 use entities::models::{BufferedTransaction, SignatureWithSlot};
+use flatbuffers::FlatBufferBuilder;
+use futures::{stream, StreamExt, TryStreamExt};
 use interface::error::UsecaseError;
-use interface::solana_rpc::{GetSignaturesByAddress, GetTransactionsBySignatures};
+use interface::solana_rpc::TransactionsGetter;
+use plerkle_serialization::serializer::seralize_encoded_transaction_with_status;
 use solana_client::nonblocking::rpc_client::RpcClient;
 use solana_client::rpc_client::GetConfirmedSignaturesForAddress2Config;
+use solana_client::rpc_config::RpcTransactionConfig;
 use solana_program::pubkey::Pubkey;
 use solana_sdk::commitment_config::{CommitmentConfig, CommitmentLevel};
 use solana_sdk::signature::Signature;
+use solana_transaction_status::UiTransactionEncoding;
 use std::str::FromStr;
+use std::sync::Arc;
 
 pub struct BackfillRPC {
-    client: RpcClient,
+    client: Arc<RpcClient>,
 }
 
 impl BackfillRPC {
     pub fn connect(addr: String) -> Self {
         Self {
-            client: RpcClient::new(addr),
+            client: Arc::new(RpcClient::new(addr)),
         }
     }
 }
 
 #[async_trait]
-impl GetSignaturesByAddress for BackfillRPC {
+impl TransactionsGetter for BackfillRPC {
     async fn get_signatures_by_address(
         &self,
         until: Signature,
@@ -53,20 +59,50 @@ impl GetSignaturesByAddress for BackfillRPC {
             })
             .collect::<Result<Vec<_>, UsecaseError>>()?)
     }
-}
 
-#[async_trait]
-impl GetTransactionsBySignatures for BackfillRPC {
     async fn get_txs_by_signatures(
         &self,
-        _signatures: Vec<Signature>,
+        signatures: Vec<Signature>,
     ) -> Result<Vec<BufferedTransaction>, UsecaseError> {
-        todo!()
+        stream::iter(signatures)
+            .map(|signature| {
+                let client = self.client.clone();
+                async move {
+                    client
+                        .get_transaction_with_config(
+                            &signature,
+                            RpcTransactionConfig {
+                                encoding: Some(UiTransactionEncoding::Base64),
+                                commitment: Some(CommitmentConfig {
+                                    commitment: CommitmentLevel::Finalized,
+                                }),
+                                max_supported_transaction_version: Some(0),
+                            },
+                        )
+                        .await
+                        .map_err(Into::<UsecaseError>::into)
+                        .and_then(|transaction| {
+                            seralize_encoded_transaction_with_status(
+                                FlatBufferBuilder::new(),
+                                transaction,
+                            )
+                            .map(|fb| BufferedTransaction {
+                                transaction: fb.finished_data().to_vec(),
+                                map_flatbuffer: false,
+                            })
+                            .map_err(Into::<UsecaseError>::into)
+                        })
+                }
+            })
+            .buffered(500) // max count of simultaneous requests
+            .try_collect::<Vec<_>>()
+            .await
     }
 }
 
+#[cfg(feature = "rpc_tests")]
 #[tokio::test]
-async fn test_get_signatures_by_address() {
+async fn test_rpc_get_signatures_by_address() {
     let client = BackfillRPC::connect("https://api.mainnet-beta.solana.com".to_string());
     let signatures = client
         .get_signatures_by_address(
@@ -79,3 +115,38 @@ async fn test_get_signatures_by_address() {
 
     assert_eq!(signatures.len(), 1000)
 }
+
+#[cfg(feature = "rpc_tests")]
+#[tokio::test]
+async fn test_rpc_get_txs_by_signatures() {
+    let client = BackfillRPC::connect("https://docs-demo.solana-mainnet.quiknode.pro/".to_string());
+    let signatures = vec![
+            Signature::from_str("2H4c1LcgWG2VuxE4rb318spyiMe1Aet5AysQHAB3Pm3z9nadxJH4C1GZD8yMeAgjdzojmLZGQppuiZqG2oKrtwF2").unwrap(),
+            Signature::from_str("265JP2HS6DwJPS4Htk2msUbxbrdeHLFVXUTFVSZ7CyMrHM8xXTxZJpLpt67kKHPAUVtEj7i3fWb5Z9vqMHREHmVm").unwrap(),
+        ];
+
+    let txs = client.get_txs_by_signatures(signatures).await.unwrap();
+
+    let parsed_txs = txs
+        .iter()
+        .map(|tx| {
+            plerkle_serialization::root_as_transaction_info(tx.transaction.as_slice()).unwrap()
+        })
+        .collect::<Vec<_>>();
+
+    assert_eq!(parsed_txs.len(), 2);
+    assert_eq!(parsed_txs[0].signature(), Some("2H4c1LcgWG2VuxE4rb318spyiMe1Aet5AysQHAB3Pm3z9nadxJH4C1GZD8yMeAgjdzojmLZGQppuiZqG2oKrtwF2"));
+    assert_eq!(parsed_txs[1].slot(), 240869063)
+}
+
+#[cfg(feature = "rpc_tests")]
+#[tokio::test]
+#[should_panic]
+async fn test_rpc_get_txs_by_signatures_error() {
+    let client = BackfillRPC::connect("https://docs-demo.solana-mainnet.quiknode.pro/".to_string());
+    let signatures = vec![
+        Signature::from_str("2H4c1LcgWG2VuxE4rb318spyiMe1Aet5AysQHAB3Pm3z9nadxJH4C1GZD8yMeAgjdzojmLZGQppuiZqG2oKrtwF3").unwrap(), // transaction that does not exists
+    ];
+
+    client.get_txs_by_signatures(signatures).await.unwrap();
+}

interface/Cargo.toml

@@ -12,4 +12,5 @@ mockall = "0.12.0"
 thiserror = "1.0.51"
 entities = { path = "../entities" }
 solana-sdk = "~1.16"
-solana-client = "~1.16"
+solana-client = "~1.16"
+plerkle_serialization = "1.5.2"

interface/src/error.rs

@@ -1,3 +1,4 @@
+use plerkle_serialization::error::PlerkleSerializationError;
 use solana_client::client_error::ClientError;
 use solana_sdk::signature::ParseSignatureError;
 use thiserror::Error;
@@ -15,6 +16,8 @@ pub enum UsecaseError {
     SolanaRPC(String),
     #[error("ParseSignature {0}")]
     ParseSignature(#[from] ParseSignatureError),
+    #[error("PlerkleSerialization {0}")]
+    PlerkleSerialization(#[from] PlerkleSerializationError),
 }
 
 impl From<ClientError> for UsecaseError {

interface/src/solana_rpc.rs

@@ -7,18 +7,13 @@ use solana_sdk::signature::Signature;
 
 #[automock]
 #[async_trait]
-pub trait GetSignaturesByAddress: Send + Sync {
+pub trait TransactionsGetter: Send + Sync {
     async fn get_signatures_by_address(
         &self,
         until: Signature,
         before: Option<Signature>,
         address: Pubkey,
     ) -> Result<Vec<SignatureWithSlot>, UsecaseError>;
-}
-
-#[automock]
-#[async_trait]
-pub trait GetTransactionsBySignatures: Send + Sync {
     async fn get_txs_by_signatures(
         &self,
         signatures: Vec<Signature>,