feat(solana-indexer) PR 15.6: adopt cow-solana-rpc

crates/cow-solana-rpc/Cargo.toml

@@ -10,9 +10,12 @@ name = "cow_solana_rpc"
 path = "src/lib.rs"
 
 [dependencies]
+futures = { workspace = true }
+itertools = { workspace = true }
 solana-commitment-config = { workspace = true }
 solana-rpc-client = { workspace = true }
 solana-rpc-client-api = { workspace = true }
+solana-sdk = { workspace = true }
 url = { workspace = true }
 
 [features]

crates/cow-solana-rpc/src/lib.rs

@@ -1,7 +1,17 @@
 //! Solana JSON-RPC client wrapper.
 
+use {
+    futures::{TryFutureExt, future},
+    itertools::Itertools,
+    solana_rpc_client::nonblocking::rpc_client::RpcClient,
+    solana_rpc_client_api::request::MAX_MULTIPLE_ACCOUNTS,
+    solana_sdk::{account::Account, pubkey::Pubkey},
+    std::{collections::HashMap, time::Duration},
+    url::Url,
+};
 pub use {solana_commitment_config::CommitmentConfig, solana_rpc_client_api::client_error::Error};
-use {solana_rpc_client::nonblocking::rpc_client::RpcClient, std::time::Duration, url::Url};
+#[cfg(feature = "test-util")]
+pub use {solana_rpc_client::mock_sender::Mocks, solana_rpc_client_api::request::RpcRequest};
 
 pub struct SolanaRPC {
     inner: RpcClient,
@@ -33,6 +43,38 @@ impl SolanaRPC {
         }
     }
 
+    /// Creates a client answering from canned per-request responses, for
+    /// tests.
+    #[cfg(feature = "test-util")]
+    pub fn new_mock_with_mocks(mocks: Mocks) -> Self {
+        Self {
+            inner: RpcClient::new_mock_with_mocks("mock".to_owned(), mocks),
+        }
+    }
+
+    /// Fetch accounts by key. Accounts that do not exist are absent from the
+    /// map. Duplicate keys are fetched once, and batches above the server's
+    /// per-request cap are split into parallel requests.
+    pub async fn multiple_accounts(
+        &self,
+        keys: impl IntoIterator<Item = Pubkey>,
+    ) -> Result<HashMap<Pubkey, Account>, Error> {
+        let unique: Vec<Pubkey> = keys.into_iter().unique().collect();
+        let fetched = future::try_join_all(unique.chunks(MAX_MULTIPLE_ACCOUNTS).map(|chunk| {
+            self.inner
+                .get_multiple_accounts(chunk)
+                .map_ok(move |accounts| {
+                    accounts
+                        .into_iter()
+                        .zip(chunk)
+                        .filter_map(|(account, key)| Some((*key, account?)))
+                        .collect::<Vec<_>>()
+                })
+        }))
+        .await?;
+        Ok(fetched.into_iter().flatten().collect())
+    }
+
     /// The node's current slot at the client's commitment level.
     pub async fn slot(&self) -> Result<u64, Error> {
         self.inner.get_slot().await

crates/solana-indexer/Cargo.toml

@@ -21,10 +21,10 @@ bigdecimal = { workspace = true }
 bytes = { workspace = true }
 clap = { workspace = true }
 configs = { workspace = true }
+cow-solana-rpc = { workspace = true }
 derive_more = { workspace = true }
 futures = { workspace = true }
 humantime-serde = { workspace = true }
-itertools = { workspace = true }
 observe = { workspace = true }
 rustls = { workspace = true }
 serde = { workspace = true }
@@ -34,9 +34,6 @@ settlement-interface = {
   git = "https://github.com/cowprotocol/solana-programs",
   rev = "4d4fc46f594e0159246afaf148f0af2a463d0095"
 }
-solana-client = { workspace = true }
-solana-commitment-config = { workspace = true }
-solana-rpc-client-api = { workspace = true }
 solana-sdk = { workspace = true }
 spl-token-2022-interface = { workspace = true }
 spl-token-interface = { workspace = true }
@@ -51,6 +48,7 @@ yellowstone-grpc-client = { workspace = true }
 yellowstone-grpc-proto = { workspace = true }
 
 [dev-dependencies]
+cow-solana-rpc = { workspace = true, features = ["test-util"] }
 serde_json = { workspace = true }
 settlement-client = {
   package = "settlement-client",

crates/solana-indexer/src/indexer/decoder.rs

@@ -7,7 +7,6 @@
 use {
     crate::{
         persistence::Postgres,
-        rpc::Rpc,
         types::{
             Signature,
             channel::StreamUpdate,
@@ -27,6 +26,7 @@ use {
         },
     },
     bytes::Bytes,
+    cow_solana_rpc::SolanaRPC,
     settlement_interface::{
         Pubkey as InterfacePubkey,
         SettlementInstruction,
@@ -50,7 +50,7 @@ pub(crate) struct Decoder {
     pub persistence: Postgres,
 
     /// Account lookups for data the stream does not carry.
-    pub rpc: Rpc,
+    pub rpc: SolanaRPC,
 
     /// Incoming `StreamUpdate` from the ingester.
     pub rx: Receiver<StreamUpdate>,
@@ -67,7 +67,7 @@ impl Decoder {
     /// Construct a new decoder. The caller owns the channel capacity decision.
     pub fn new(
         persistence: Postgres,
-        rpc: Rpc,
+        rpc: SolanaRPC,
         rx: Receiver<StreamUpdate>,
         settlement_program: Pubkey,
         solflow_program: Option<Pubkey>,

crates/solana-indexer/src/indexer/decoder/tests.rs

@@ -3,7 +3,6 @@ use {
     crate::{
         indexer::ingester::Ingester,
         persistence::Postgres,
-        rpc::Rpc,
         types::{
             Signature,
             events::{
@@ -34,6 +33,7 @@ use {
         },
     },
     bytes::Bytes,
+    cow_solana_rpc::{Mocks, RpcRequest, SolanaRPC},
     futures::StreamExt,
     settlement_interface::{
         Pubkey as InterfacePubkey,
@@ -411,7 +411,7 @@ fn token_account_mint_trusts_only_token_program_accounts() {
 /// A canned `getMultipleAccounts` response, in request order: the sell token
 /// account holds mint `[0xA1; 32]`, the buy token account mint `[0xA2; 32]`
 /// (the first 32 bytes of the base64 data).
-fn mock_rpc_with_token_accounts() -> Rpc {
+fn mock_rpc_with_token_accounts() -> SolanaRPC {
     let account = |data: &str| {
         serde_json::json!({
             "lamports": 1u64,
@@ -428,11 +428,8 @@ fn mock_rpc_with_token_accounts() -> Rpc {
         "context": { "slot": 1u64, "apiVersion": "2.0.0" },
         "value": [sell, buy],
     });
-    let mocks = solana_client::rpc_client::Mocks::from([(
-        solana_client::rpc_request::RpcRequest::GetMultipleAccounts,
-        response,
-    )]);
-    Rpc::new_mock(mocks)
+    let mocks = Mocks::from([(RpcRequest::GetMultipleAccounts, response)]);
+    SolanaRPC::new_mock_with_mocks(mocks)
 }
 
 /// A decoder over a lazy pool that never connects: `decode` is pure, tests
@@ -442,7 +439,7 @@ fn pure_decoder(settlement: Pubkey, solflow: Pubkey) -> Decoder {
     let (_sender, rx) = tokio::sync::mpsc::channel(1);
     Decoder::new(
         Postgres::new(pool),
-        Rpc::new_mock(Default::default()),
+        SolanaRPC::new_mock_with_mocks(Default::default()),
         rx,
         settlement,
         Some(solflow),

crates/solana-indexer/src/lib.rs

@@ -5,7 +5,6 @@
 pub mod config;
 pub mod indexer;
 pub mod persistence;
-pub mod rpc;
 pub mod run;
 #[cfg(test)]
 mod test_db;

crates/solana-indexer/src/run.rs

@@ -9,10 +9,10 @@ use {
             ingester::{Error, INGEST_TO_DECODER_CAPACITY, Ingester, Resume},
         },
         persistence::Postgres,
-        rpc::Rpc,
         yellowstone,
     },
     clap::Parser,
+    cow_solana_rpc::{CommitmentConfig, SolanaRPC},
     sqlx::postgres::PgPoolOptions,
     std::{
         path::PathBuf,
@@ -61,7 +61,12 @@ async fn run(config: Config, start_slot: Option<u64>) {
         .await
         .expect("database connection");
     let persistence = Postgres::new(pool);
-    let rpc = Rpc::new(config.rpc.endpoint, config.rpc.request_timeout);
+    // Confirmed commitment, matching the stream subscription.
+    let rpc = SolanaRPC::new_with_timeout_and_commitment(
+        &config.rpc.endpoint,
+        config.rpc.request_timeout,
+        CommitmentConfig::confirmed(),
+    );
 
     let (tx, rx) = mpsc::channel(INGEST_TO_DECODER_CAPACITY);
     let settlement_program = config.chain.settlement_program_id;

crates/solana-indexer/src/types/errors.rs

@@ -38,7 +38,7 @@ pub(crate) enum PersistenceError {
     Database(#[from] sqlx::Error),
     /// The account lookup enriching a batch failed.
     #[error("rpc: {0}")]
-    Rpc(#[from] solana_client::client_error::ClientError),
+    Rpc(#[from] cow_solana_rpc::Error),
 }
 
 /// Failures surfaced from the stream boundary.