2.0.16

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "geyser-grpc-connector"
-version = "0.10.1+yellowstone.1.12"
+version = "2.0.0"
 edition = "2021"
 
 description = "Multiplexing and Reconnection on Yellowstone gRPC Geyser client streaming"
@@ -9,12 +9,12 @@ authors = ["GroovieGermanikus <[email protected]>"]
 repository = "https://github.com/blockworks-foundation/geyser-grpc-connector"
 
 [dependencies]
-yellowstone-grpc-client = { version = "1.13.0+solana.1.17.15", git = "https://github.com/rpcpool/yellowstone-grpc.git", tag = "v1.12.0+solana.1.17.15" }
-yellowstone-grpc-proto = { version = "1.12.0+solana.1.17.15", git = "https://github.com/rpcpool/yellowstone-grpc.git", tag = "v1.12.0+solana.1.17.15" }
+yellowstone-grpc-client = { version = "2.0.0", git = "https://github.com/rpcpool/yellowstone-grpc.git", tag = "v2.0.0+solana.2.0.16" }
+yellowstone-grpc-proto = { version = "2.0.0", git = "https://github.com/rpcpool/yellowstone-grpc.git", tag = "v2.0.0+solana.2.0.16" }
 
 
 # required for CommitmentConfig
-solana-sdk = "~1.17.15"
+solana-sdk = "2.0.16"
 
 url = "2.5.0"
 async-stream = "0.3.5"
@@ -32,6 +32,10 @@ bincode = "1.3.3"
 
 csv = "1.3.0"
 
-[dev-dependencies]
+#[dev-dependencies]
 tracing-subscriber = "0.3.16"
-solana-logger = "1"
+solana-logger = "2.0.16"
+
+[patch.crates-io.curve25519-dalek]
+git = "https://github.com/anza-xyz/curve25519-dalek.git"
+rev = "b500cdc2a920cd5bff9e2dd974d7b97349d61464"

README.md

@@ -23,7 +23,7 @@ These are the currently maintained versions of the library: [see Wiki](https://g
 ```cargo add geyser-grpc-connector ```
 
 
-An example how to use the library is provided in `stream_blocks_mainnet.rs`.
+An example how to use the library is provided in `stream_blocks_mainnet_stream.rs`.
 
 ## Known issues
 * Library does not support other data than Blocks/Slots very well.

ci/rust-version.sh

@@ -23,7 +23,7 @@
 if [[ -n $RUST_STABLE_VERSION ]]; then
   stable_version="$RUST_STABLE_VERSION"
 else
-  stable_version=1.70.0
+  stable_version=1.81.0
 fi
 
 if [[ -n $RUST_NIGHTLY_VERSION ]]; then
@@ -91,7 +91,7 @@ export rust_nightly_docker_image=solanalabs/rust-nightly:"$nightly_version"
 if [[ -n $RUST_STABLE_VERSION ]]; then
   stable_version="$RUST_STABLE_VERSION"
 else
-  stable_version=1.70.0
+  stable_version=1.81.0
 fi
 
 if [[ -n $RUST_NIGHTLY_VERSION ]]; then

examples/dump_slots_stream_samples.rs

@@ -7,20 +7,19 @@
 ///
 /// we want to see if there is a difference in timing of "processed accounts" in the mix with slot vs "only processed accounts"
 use log::{info, warn};
 use solana_sdk::clock::Slot;
 use solana_sdk::commitment_config::{CommitmentConfig, CommitmentLevel};
 use std::collections::HashMap;
 use std::env;
 use std::str::FromStr;
 use std::time::SystemTime;
 
 use base64::Engine;
 use csv::ReaderBuilder;
 use itertools::Itertools;
-use solana_sdk::borsh0_10::try_from_slice_unchecked;
 /// This file mocks the core model of the RPC server.
 use solana_sdk::compute_budget;
 use solana_sdk::compute_budget::ComputeBudgetInstruction;
 use solana_sdk::hash::Hash;
 use solana_sdk::instruction::CompiledInstruction;
 use solana_sdk::message::v0::MessageAddressTableLookup;
@@ -167,7 +166,7 @@
         .expect("valid commitment level")
 }
 
-#[tokio::main]
+#[tokio::main(flavor = "current_thread")]
 pub async fn main() {
     tracing_subscriber::fmt::init();
 

examples/stream_blocks_autoconnect.rs

@@ -60,7 +60,7 @@ enum TestCases {
 }
 const TEST_CASE: TestCases = TestCases::Basic;
 
-#[tokio::main]
+#[tokio::main(flavor = "current_thread")]
 pub async fn main() {
     // RUST_LOG=info,stream_blocks_mainnet=debug,geyser_grpc_connector=trace
     tracing_subscriber::fmt::init();

examples/stream_blocks_mainnet_stream.rs

@@ -8,14 +8,14 @@ use std::pin::pin;
 use base64::Engine;
 use itertools::Itertools;
 use solana_sdk::borsh0_10::try_from_slice_unchecked;
-/// This file mocks the core model of the RPC server.
-use solana_sdk::compute_budget;
 use solana_sdk::compute_budget::ComputeBudgetInstruction;
 use solana_sdk::hash::Hash;
 use solana_sdk::instruction::CompiledInstruction;
 use solana_sdk::message::v0::MessageAddressTableLookup;
 use solana_sdk::message::{v0, MessageHeader, VersionedMessage};
 use solana_sdk::pubkey::Pubkey;
+/// This file mocks the core model of the RPC server.
+use solana_sdk::{borsh1, compute_budget};
 
 use solana_sdk::signature::Signature;
 use solana_sdk::transaction::TransactionError;
@@ -97,7 +97,7 @@ impl FromYellowstoneExtractor for BlockMetaExtractor {
     }
 }
 
-#[tokio::main]
+#[tokio::main(flavor = "current_thread")]
 pub async fn main() {
     // RUST_LOG=info,stream_blocks_mainnet=debug,geyser_grpc_connector=trace
     tracing_subscriber::fmt::init();
@@ -298,66 +298,32 @@ pub fn map_produced_block(
                     .collect(),
             });
 
-            let legacy_compute_budget: Option<(u32, Option<u64>)> =
-                message.instructions().iter().find_map(|i| {
-                    if i.program_id(message.static_account_keys())
-                        .eq(&compute_budget::id())
-                    {
-                        if let Ok(ComputeBudgetInstruction::RequestUnitsDeprecated {
-                            units,
-                            additional_fee,
-                        }) = try_from_slice_unchecked(i.data.as_slice())
-                        {
-                            if additional_fee > 0 {
-                                return Some((
-                                    units,
-                                    Some(((units * 1000) / additional_fee) as u64),
-                                ));
-                            } else {
-                                return Some((units, None));
-                            }
-                        }
-                    }
-                    None
-                });
-
-            let legacy_cu_requested = legacy_compute_budget.map(|x| x.0);
-            let legacy_prioritization_fees = legacy_compute_budget.map(|x| x.1).unwrap_or(None);
-
-            let cu_requested = message
-                .instructions()
-                .iter()
-                .find_map(|i| {
-                    if i.program_id(message.static_account_keys())
-                        .eq(&compute_budget::id())
+            let cu_requested = message.instructions().iter().find_map(|i| {
+                if i.program_id(message.static_account_keys())
+                    .eq(&compute_budget::id())
+                {
+                    if let Ok(ComputeBudgetInstruction::SetComputeUnitLimit(limit)) =
+                        borsh1::try_from_slice_unchecked(i.data.as_slice())
                     {
-                        if let Ok(ComputeBudgetInstruction::SetComputeUnitLimit(limit)) =
-                            try_from_slice_unchecked(i.data.as_slice())
-                        {
-                            return Some(limit);
-                        }
+                        return Some(limit);
                     }
-                    None
-                })
-                .or(legacy_cu_requested);
-
-            let prioritization_fees = message
-                .instructions()
-                .iter()
-                .find_map(|i| {
-                    if i.program_id(message.static_account_keys())
-                        .eq(&compute_budget::id())
+                }
+                None
+            });
+
+            let prioritization_fees = message.instructions().iter().find_map(|i| {
+                if i.program_id(message.static_account_keys())
+                    .eq(&compute_budget::id())
+                {
+                    if let Ok(ComputeBudgetInstruction::SetComputeUnitPrice(price)) =
+                        borsh1::try_from_slice_unchecked(i.data.as_slice())
                     {
-                        if let Ok(ComputeBudgetInstruction::SetComputeUnitPrice(price)) =
-                            try_from_slice_unchecked(i.data.as_slice())
-                        {
-                            return Some(price);
-                        }
+                        return Some(price);
                     }
+                }
 
-                    None
-                })
-                .or(legacy_prioritization_fees);
+                None
+            });
 
             Some(TransactionInfo {
                 signature: signature.to_string(),

examples/stream_blocks_mainnet_task.rs

@@ -1,21 +1,21 @@
 use futures::Stream;
 use log::{info, warn};
 use solana_sdk::clock::Slot;
 use solana_sdk::commitment_config::CommitmentConfig;
 use std::env;
 use std::pin::pin;
 
 use base64::Engine;
 use itertools::Itertools;
 use solana_sdk::borsh0_10::try_from_slice_unchecked;
-/// This file mocks the core model of the RPC server.
-use solana_sdk::compute_budget;
 use solana_sdk::compute_budget::ComputeBudgetInstruction;
 use solana_sdk::hash::Hash;
 use solana_sdk::instruction::CompiledInstruction;
 use solana_sdk::message::v0::MessageAddressTableLookup;
 use solana_sdk::message::{v0, MessageHeader, VersionedMessage};
 use solana_sdk::pubkey::Pubkey;
+/// This file mocks the core model of the RPC server.
+use solana_sdk::{borsh1, compute_budget};
 
 use solana_sdk::signature::Signature;
 use solana_sdk::transaction::TransactionError;
@@ -23,10 +23,10 @@
 use yellowstone_grpc_proto::geyser::SubscribeUpdateBlock;
 
 use geyser_grpc_connector::grpc_subscription_autoreconnect_tasks::{
     create_geyser_autoconnection_task, create_geyser_autoconnection_task_with_mpsc,
 };
 use geyser_grpc_connector::grpcmultiplex_fastestwins::{
     create_multiplexed_stream, FromYellowstoneExtractor,
 };
 use geyser_grpc_connector::{GeyserFilter, GrpcConnectionTimeouts, GrpcSourceConfig, Message};
 use tokio::time::{sleep, Duration};
@@ -93,7 +93,7 @@
     }
 }
 
-#[tokio::main]
+#[tokio::main(flavor = "current_thread")]
 pub async fn main() {
     // RUST_LOG=info,stream_blocks_mainnet=debug,geyser_grpc_connector=trace
     tracing_subscriber::fmt::init();
@@ -268,66 +268,32 @@
                     .collect(),
             });
 
-            let legacy_compute_budget: Option<(u32, Option<u64>)> =
-                message.instructions().iter().find_map(|i| {
-                    if i.program_id(message.static_account_keys())
-                        .eq(&compute_budget::id())
+            let cu_requested = message.instructions().iter().find_map(|i| {
+                if i.program_id(message.static_account_keys())
+                    .eq(&compute_budget::id())
+                {
+                    if let Ok(ComputeBudgetInstruction::SetComputeUnitLimit(limit)) =
+                        borsh1::try_from_slice_unchecked(i.data.as_slice())
                     {
-                        if let Ok(ComputeBudgetInstruction::RequestUnitsDeprecated {
-                            units,
-                            additional_fee,
-                        }) = try_from_slice_unchecked(i.data.as_slice())
-                        {
-                            if additional_fee > 0 {
-                                return Some((
-                                    units,
-                                    Some(((units * 1000) / additional_fee) as u64),
-                                ));
-                            } else {
-                                return Some((units, None));
-                            }
-                        }
+                        return Some(limit);
                     }
-                    None
-                });
-
-            let legacy_cu_requested = legacy_compute_budget.map(|x| x.0);
-            let legacy_prioritization_fees = legacy_compute_budget.map(|x| x.1).unwrap_or(None);
-
-            let cu_requested = message
-                .instructions()
-                .iter()
-                .find_map(|i| {
-                    if i.program_id(message.static_account_keys())
-                        .eq(&compute_budget::id())
-                    {
-                        if let Ok(ComputeBudgetInstruction::SetComputeUnitLimit(limit)) =
-                            try_from_slice_unchecked(i.data.as_slice())
-                        {
-                            return Some(limit);
-                        }
-                    }
-                    None
-                })
-                .or(legacy_cu_requested);
-
-            let prioritization_fees = message
-                .instructions()
-                .iter()
-                .find_map(|i| {
-                    if i.program_id(message.static_account_keys())
-                        .eq(&compute_budget::id())
+                }
+                None
+            });
+
+            let prioritization_fees = message.instructions().iter().find_map(|i| {
+                if i.program_id(message.static_account_keys())
+                    .eq(&compute_budget::id())
+                {
+                    if let Ok(ComputeBudgetInstruction::SetComputeUnitPrice(price)) =
+                        borsh1::try_from_slice_unchecked(i.data.as_slice())
                     {
-                        if let Ok(ComputeBudgetInstruction::SetComputeUnitPrice(price)) =
-                            try_from_slice_unchecked(i.data.as_slice())
-                        {
-                            return Some(price);
-                        }
+                        return Some(price);
                     }
+                }
 
-                    None
-                })
-                .or(legacy_prioritization_fees);
+                None
+            });
 
             Some(TransactionInfo {
                 signature: signature.to_string(),

examples/stream_blocks_single.rs

@@ -2,10 +2,11 @@ use futures::{Stream, StreamExt};
 use log::info;
 use solana_sdk::clock::Slot;
 use solana_sdk::commitment_config::CommitmentConfig;
+use solana_sdk::pubkey::Pubkey;
 use std::env;
 use std::pin::pin;
-use solana_sdk::pubkey::Pubkey;
 
+use csv::Writer;
 use geyser_grpc_connector::grpc_subscription_autoreconnect_streams::create_geyser_reconnecting_stream;
 use geyser_grpc_connector::grpcmultiplex_fastestwins::FromYellowstoneExtractor;
 use geyser_grpc_connector::{GeyserFilter, GrpcConnectionTimeouts, GrpcSourceConfig, Message};
@@ -14,7 +15,6 @@ use tracing::warn;
 use yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof;
 use yellowstone_grpc_proto::geyser::SubscribeUpdate;
 use yellowstone_grpc_proto::prost::Message as _;
-use csv::Writer;
 
 #[allow(dead_code)]
 fn start_example_blockmini_consumer(
@@ -68,7 +68,7 @@ impl FromYellowstoneExtractor for BlockMiniExtractor {
     }
 }
 
-#[tokio::main]
+#[tokio::main(flavor = "current_thread")]
 pub async fn main() {
     // RUST_LOG=info,stream_blocks_mainnet=debug,geyser_grpc_connector=trace
     tracing_subscriber::fmt::init();
@@ -114,10 +114,13 @@ pub async fn main() {
                         Some(UpdateOneof::Account(update)) => {
                             let account_info = update.account.unwrap();
                             let account_pk = Pubkey::try_from(account_info.pubkey).unwrap();
-                            info!("got account update (green)!!! {} - {:?} - {} bytes",
-                                update.slot, account_pk, account_info.data.len());
-                            let bytes: [u8; 32] =
-                                account_pk.to_bytes();
+                            info!(
+                                "got account update (green)!!! {} - {:?} - {} bytes",
+                                update.slot,
+                                account_pk,
+                                account_info.data.len()
+                            );
+                            let bytes: [u8; 32] = account_pk.to_bytes();
                         }
                         _ => {}
                     }
@@ -138,7 +141,10 @@ pub async fn main() {
                 Message::GeyserSubscribeUpdate(subscriber_update) => {
                     let mapped = extractor.map_yellowstone_update(*subscriber_update);
                     if let Some((slot, block_mini)) = mapped {
-                        info!("got update (blue)!!! block: {} - {} bytes", slot, block_mini.blocksize);
+                        info!(
+                            "got update (blue)!!! block: {} - {} bytes",
+                            slot, block_mini.blocksize
+                        );
                     }
                 }
                 Message::Connecting(attempt) => {

rust-toolchain.toml

@@ -1,2 +1,2 @@
 [toolchain]
-channel = "1.73.0"
+channel = "1.81.0"