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,15 +1,19 @@
 use async_trait::async_trait;
 use entities::models::BufferedTransaction;
+use flatbuffers::FlatBufferBuilder;
+use futures::{stream, StreamExt};
 use interface::error::UsecaseError;
-use interface::solana_rpc::{
-    GetSignaturesByAddress, GetTransactionsBySignatures, SignatureWithSlot,
-};
+use interface::solana_rpc::{GetBackfillTransactions, SignatureWithSlot};
+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,
@@ -24,7 +28,7 @@ impl BackfillRPC {
 }
 
 #[async_trait]
-impl GetSignaturesByAddress for BackfillRPC {
+impl GetBackfillTransactions for BackfillRPC {
     async fn get_signatures_by_address(
         &self,
         until: Signature,
@@ -55,20 +59,48 @@ impl GetSignaturesByAddress for BackfillRPC {
             })
             .collect::<Result<Vec<_>, UsecaseError>>()?)
     }
-}
 
-#[async_trait]
-impl GetTransactionsBySignatures for BackfillRPC {
-    async fn get_txs_by_signatures(
-        &self,
-        _signatures: Vec<Signature>,
-    ) -> Result<Vec<BufferedTransaction>, UsecaseError> {
-        todo!()
+    async fn get_txs_by_signatures(&self, signatures: Vec<Signature>) -> Vec<BufferedTransaction> {
+        let client = Arc::new(&self.client);
+
+        stream::iter(signatures)
+            .map(|signature| {
+                let client = client.clone();
+                async move {
+                    let transaction = client
+                        .get_transaction_with_config(
+                            &signature,
+                            RpcTransactionConfig {
+                                encoding: Some(UiTransactionEncoding::Base64),
+                                commitment: Some(CommitmentConfig {
+                                    commitment: CommitmentLevel::Finalized,
+                                }),
+                                max_supported_transaction_version: Some(0),
+                            },
+                        )
+                        .await
+                        .ok()?;
+
+                    let builder = FlatBufferBuilder::new();
+                    Some(BufferedTransaction {
+                        transaction: seralize_encoded_transaction_with_status(builder, transaction)
+                            .ok()?
+                            .finished_data()
+                            .to_vec(),
+                        map_flatbuffer: false,
+                    })
+                }
+            })
+            .buffered(500) // max count of simultaneous requests
+            .filter_map(|transaction| async { transaction }) // skip all None results
+            .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(
@@ -81,3 +113,28 @@ 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("2H4c1LcgWG2VuxE4rb318spyiMe1Aet5AysQHAB3Pm3z9nadxJH4C1GZD8yMeAgjdzojmLZGQppuiZqG2oKrtwF1").unwrap(), // transaction that does not exists
+            Signature::from_str("2H4c1LcgWG2VuxE4rb318spyiMe1Aet5AysQHAB3Pm3z9nadxJH4C1GZD8yMeAgjdzojmLZGQppuiZqG2oKrtwF2").unwrap(),
+            Signature::from_str("2H4c1LcgWG2VuxE4rb318spyiMe1Aet5AysQHAB3Pm3z9nadxJH4C1GZD8yMeAgjdzojmLZGQppuiZqG2oKrtwF3").unwrap(), // transaction that does not exists
+            Signature::from_str("265JP2HS6DwJPS4Htk2msUbxbrdeHLFVXUTFVSZ7CyMrHM8xXTxZJpLpt67kKHPAUVtEj7i3fWb5Z9vqMHREHmVm").unwrap(),
+        ];
+
+    let txs = client.get_txs_by_signatures(signatures).await;
+
+    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)
+}

interface/src/solana_rpc.rs

@@ -14,20 +14,12 @@ pub struct SignatureWithSlot {
 
 #[automock]
 #[async_trait]
-pub trait GetSignaturesByAddress: Send + Sync {
+pub trait GetBackfillTransactions: 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>,
-    ) -> Result<Vec<BufferedTransaction>, UsecaseError>;
+    async fn get_txs_by_signatures(&self, signatures: Vec<Signature>) -> Vec<BufferedTransaction>;
 }