Initial changes for migrating to lite-rpc-0.23.0

Cargo.toml

@@ -34,21 +34,22 @@ thiserror = "1.0"
 tokio = { version = "1.14.1", features = ["full"] }
 regex = "1.7.3"
 
-solana-client = "1.15.2"
-solana-metrics = "1.15.2"
-solana-rpc-client = "1.15.2"
-solana-runtime = "1.15.2"
-solana-sdk = "1.15.2"
-solana-clap-utils = "1.15.2"
-solana-cli-config = "1.15.2"
-solana-net-utils = "1.15.2"
-solana-version = "1.15.2"
-solana-logger = "1.15.2"
-solana-transaction-status = "1.15.2"
-solana-account-decoder = "1.15.2"
+solana-client = "1.16.13"
+solana-metrics = "1.16.13"
+solana-rpc-client = "1.16.13"
+solana-runtime = "1.16.13"
+solana-sdk = "1.16.13"
+solana-clap-utils = "1.16.13"
+solana-cli-config = "1.16.13"
+solana-net-utils = "1.16.13"
+solana-version = "1.16.13"
+solana-logger = "1.16.13"
+solana-transaction-status = "1.16.13"
+solana-account-decoder = "1.16.13"
 
-solana-lite-rpc-core = { git = "https://github.com/blockworks-foundation/lite-rpc.git", tag = "v0.2.2" }
-solana-lite-rpc-services = { git = "https://github.com/blockworks-foundation/lite-rpc.git", tag = "v0.2.2" }
+solana-lite-rpc-core = { git = "https://github.com/blockworks-foundation/lite-rpc.git", tag = "v0.2.3" }
+solana-lite-rpc-services = { git = "https://github.com/blockworks-foundation/lite-rpc.git", tag = "v0.2.3" }
+solana-lite-rpc-cluster-endpoints = { git = "https://github.com/blockworks-foundation/lite-rpc.git", tag = "v0.2.3" }
 
 
 # pin program to mango-v3 version of solana sdk
@@ -58,9 +59,9 @@ solana-program = "1.9.17"
 
 mango = { git = "https://github.com/blockworks-foundation/mango-v3.git", tag = "v3.6.0", default-features = false }
 mango-common = { git = "https://github.com/blockworks-foundation/mango-v3.git", tag = "v3.6.0" }
-mango-feeds-connector = { git = "https://github.com/blockworks-foundation/mango-feeds.git", branch = "ckamm/solana-versions2", default-features = false, features = ["solana-1-15"] }
+mango-feeds-connector = { git = "https://github.com/blockworks-foundation/mango-feeds.git", branch = "main", default-features = false, features = ["solana-1-15"] }
 bincode = "1.3.3"
-"yellowstone-grpc-proto" = "=1.1.0"
+yellowstone-grpc-proto = "1.9.0"
 
 [package.metadata.docs.rs]
 targets = ["x86_64-unknown-linux-gnu"]

rust-toolchain.toml

@@ -1,2 +1,2 @@
 [toolchain]
-channel = "1.67.1"
+channel = "1.70.0"

src/confirmation_strategies.rs

@@ -10,7 +10,7 @@ use chrono::Utc;
 use dashmap::DashMap;
 use log::{debug, warn};
 use solana_client::{nonblocking::rpc_client::RpcClient, rpc_config::RpcBlockConfig};
-use solana_lite_rpc_core::notifications::NotificationMsg;
+use solana_lite_rpc_core::structures::notifications::NotificationMsg;
 use solana_sdk::{
     commitment_config::{CommitmentConfig, CommitmentLevel},
     signature::Signature,

src/crank.rs

@@ -7,6 +7,7 @@ use crate::{
     tpu_manager::TpuManager,
 };
 
+use iter_tools::Itertools;
 use mango_feeds_connector::{
     account_write_filter::{self, AccountWriteRoute},
     metrics, websocket_source, FilterConfig, MetricsConfig, SnapshotSourceConfig, SourceConfig,
@@ -60,13 +61,13 @@ pub fn start(
     let group_pk = Pubkey::from_str(&group.public_key).unwrap();
     let cache_pk = Pubkey::from_str(&group.cache_key).unwrap();
     let mango_program_id = Pubkey::from_str(&group.mango_program_id).unwrap();
+    let accounts = group
+        .perp_markets
+        .iter()
+        .map(|m| m.events_key.clone())
+        .collect_vec();
     let filter_config = FilterConfig {
-        program_ids: vec![group.mango_program_id.clone()],
-        account_ids: group
-            .perp_markets
-            .iter()
-            .map(|m| m.events_key.clone())
-            .collect(),
+        entity_filter: mango_feeds_connector::EntityFilter::filter_by_account_ids(accounts.iter().map(|x| x.as_str()).collect()),
     };
 
     let (instruction_sender, instruction_receiver) = unbounded::<(Pubkey, Vec<Instruction>)>();
@@ -167,7 +168,6 @@ pub fn start(
                 grpc_sources: vec![],
                 snapshot: SnapshotSourceConfig {
                     rpc_http_url: config.rpc_url,
-                    program_id: config.program_id.to_string(),
                 },
                 rpc_ws_url: config.websocket_url,
             },

src/main.rs

@@ -1,3 +1,7 @@
+use solana_lite_rpc_core::{stores::{data_cache::{DataCache, SlotCache}, tx_store::{TxStore, empty_tx_store}, block_information_store::BlockInformationStore, subscription_store::SubscriptionStore, cluster_info_store::ClusterInfo}, structures::{identity_stakes::IdentityStakes, notifications::NotificationMsg}};
+use solana_lite_rpc_services::{tpu_utils::tpu_connection_path::TpuConnectionPath, data_caching_service::DataCachingService};
+use solana_sdk::signer::Signer;
+
 use {
     log::info,
     mango_simulation::{
@@ -18,13 +22,9 @@ use {
     },
     solana_client::nonblocking::rpc_client::RpcClient as NbRpcClient,
     solana_lite_rpc_core::{
-        block_store::BlockStore,
-        notifications::NotificationMsg,
         quic_connection_utils::QuicConnectionParameters,
-        tx_store::{empty_tx_store, TxStore},
     },
     solana_lite_rpc_services::{
-        block_listenser::BlockListener,
         tpu_utils::tpu_service::{TpuService, TpuServiceConfig},
         transaction_replayer::TransactionReplayer,
         transaction_service::{TransactionService, TransactionServiceBuilder},
@@ -48,18 +48,13 @@ const METRICS_NAME: &str = "mango-bencher";
 async fn configure_transaction_service(
     rpc_client: Arc<NbRpcClient>,
     identity: Keypair,
-    block_store: BlockStore,
     tx_store: TxStore,
     notifier: UnboundedSender<NotificationMsg>,
 ) -> (TransactionService, JoinHandle<anyhow::Result<()>>) {
     let slot = rpc_client.get_slot().await.expect("GetSlot should work");
     let tpu_config = TpuServiceConfig {
         fanout_slots: 12,
-        number_of_leaders_to_cache: 1024,
-        clusterinfo_refresh_time: Duration::from_secs(60 * 60),
-        leader_schedule_update_frequency: Duration::from_secs(10),
         maximum_transaction_in_queue: 200_000,
-        maximum_number_of_errors: 10,
         quic_connection_params: QuicConnectionParameters {
             connection_timeout: Duration::from_secs(1),
             connection_retry_count: 10,
@@ -69,14 +64,39 @@ async fn configure_transaction_service(
             write_timeout: Duration::from_secs(1),
             number_of_transactions_per_unistream: 10,
         },
+        tpu_connection_path: TpuConnectionPath::QuicDirectPath,
+    };
+
+    let (subscriptions, cluster_endpoint_tasks) = create_json_rpc_polling_subscription(rpc_client.clone());
+    let EndpointStreaming {
+        blocks_notifier,
+        cluster_info_notifier,
+        slot_notifier,
+        vote_account_notifier,
+    } = subscriptions;
+    let finalized_block =
+        get_latest_block(blocks_notifier.resubscribe(), CommitmentConfig::finalized()).await;
+
+
+    let data_cache = DataCache {
+        block_information_store,
+        txs: tx_store,
+        tx_subs: SubscriptionStore::default(),
+        slot_cache: SlotCache::new(finalized_block.slot),
+        identity_stakes: IdentityStakes::new(identity.pubkey()),
+        cluster_info: ClusterInfo::default(),
+    };
+
+    let lata_cache_service = DataCachingService {
+        data_cache: data_cache.clone(),
+        clean_duration: Duration::from_secs(120),
     };
 
     let tpu_service = TpuService::new(
         tpu_config,
         Arc::new(identity),
         slot,
-        rpc_client.clone(),
-        tx_store.clone(),
+        data_cache,
     )
     .await
     .expect("Should be able to create TPU");
@@ -92,7 +112,6 @@ async fn configure_transaction_service(
     let builder = TransactionServiceBuilder::new(
         tx_sender,
         replayer,
-        block_listenser,
         tpu_service,
         1_000_000,
     );
@@ -157,9 +176,6 @@ pub async fn main() -> anyhow::Result<()> {
     ));
 
     let tx_store = empty_tx_store();
-    let block_store = BlockStore::new(&nb_rpc_client)
-        .await
-        .expect("Blockstore should be created");
 
     let (notif_sx, notif_rx) = unbounded_channel();
     let (transaction_service, tx_service_jh) = configure_transaction_service(