Skip to main content

Crate yellowstone_fumarole_client

Crate yellowstone_fumarole_client 

Source
Expand description

A Rust implementation of the Yellowstone Fumarole Client using Tokio and Tonic.

Fumarole Client uses gRPC connections to communicate with the Fumarole service.

§Yellowstone-GRPC vs Yellowstone-Fumarole

For the most part, the API is similar to the original yellowstone-grpc client.

However, there are some differences:

  • The yellowstone-fumarole client can use multiple gRPC connections to communicate with the Fumarole service, helping avoid HoL blocking.
  • yellowstone-fumarole subscribers are persistent and can be reused across multiple sessions.
  • The yellowstone-fumarole can reconnect to the Fumarole service if the connection is lost.

§Examples

Examples can be found in the examples directory.

§Create a FumaroleClient

To create a FumaroleClient, you need to provide a configuration object.

ⓘ
use yellowstone_fumarole_client::FumaroleClient;
use yellowstone_fumarole_client::config::FumaroleConfig;

#[tokio::main]
async fn main() {
    let config = FumaroleConfig {
        endpoint: "https://example.com".to_string(),
        x_token: Some("00000000-0000-0000-0000-000000000000".to_string()),
        max_decoding_message_size_bytes: FumaroleConfig::default_max_decoding_message_size_bytes(),
        x_metadata: Default::default(),
    };
    let fumarole_client = FumaroleClient::connect(config)
        .await
        .expect("Failed to connect to fumarole");
}

The preferred way to create FumaroleConfig is to use serde_yaml deserialization from a YAML file.

ⓘ
let config_file = std::fs::File::open("path/to/config.yaml").unwrap();
let config: FumaroleConfig = serde_yaml::from_reader(config_file).unwrap();

Here’s an example of a YAML file:

endpoint: https://example.com
x-token: 00000000-0000-0000-0000-000000000000
response_compression: zstd

§Dragonsmouth-like Subscribe (deprecated)

use {
    clap::Parser,
    solana_sdk::{bs58, pubkey::Pubkey},
    std::{collections::HashMap, path::PathBuf},
    yellowstone_fumarole_client::{
        config::FumaroleConfig, DragonsmouthAdapterSession, FumaroleClient,
    },
    yellowstone_grpc_proto::geyser::{
        subscribe_update::UpdateOneof, SubscribeRequest,
        SubscribeRequestFilterTransactions, SubscribeUpdateAccount, SubscribeUpdateTransaction,
    },
};

#[derive(Debug, Clone, Parser)]
#[clap(author, version, about = "Yellowstone Fumarole Example")]
struct Args {
    /// Path to static config file
    #[clap(long)]
    config: PathBuf,

    #[clap(subcommand)]
    action: Action,
}

#[derive(Debug, Clone, Parser)]
enum Action {
    /// Subscribe to fumarole events
    Subscribe(SubscribeArgs),
}

#[derive(Debug, Clone, Parser)]
struct SubscribeArgs {
    /// Name of the persistent subscriber to use
    #[clap(long)]
    name: String,
}

fn summarize_account(account: SubscribeUpdateAccount) -> Option<String> {
    let slot = account.slot;
    let account = account.account?;
    let pubkey = Pubkey::try_from(account.pubkey).expect("Failed to parse pubkey");
    let owner = Pubkey::try_from(account.owner).expect("Failed to parse owner");
    Some(format!("account,{},{},{}", slot, pubkey, owner))
}

fn summarize_tx(tx: SubscribeUpdateTransaction) -> Option<String> {
    let slot = tx.slot;
    let tx = tx.transaction?;
    let sig = bs58::encode(tx.signature).into_string();
    Some(format!("tx,{slot},{sig}"))
}

async fn subscribe(args: SubscribeArgs, config: FumaroleConfig) {
    // This request listen for all account updates and transaction updates
    let request = SubscribeRequest {
        transactions: HashMap::from([(
            "f1".to_owned(),
            SubscribeRequestFilterTransactions::default(),
        )]),
        ..Default::default()
    };

    let mut fumarole_client = FumaroleClient::connect(config)
        .await
        .expect("Failed to connect to fumarole");

    let dragonsmouth_session = fumarole_client
        .dragonsmouth_subscribe(args.name, request)
        .await
        .expect("Failed to subscribe");

    let DragonsmouthAdapterSession {
        sink: _,
        mut source,
        mut fumarole_handle,
    } = dragonsmouth_session;
     
    loop {

        tokio::select! {
            result = &mut fumarole_handle => {
                eprintln!("Fumarole handle closed: {:?}", result);
                break;
            }
            maybe = source.recv() => {
                match maybe {
                    None => {
                        eprintln!("Source closed");
                        break;
                    }
                    Some(result) => {
                        let event = result.expect("Failed to receive event");
                        let message = if let Some(oneof) = event.update_oneof {
                            match oneof {
                                UpdateOneof::Account(account_update) => {
                                    summarize_account(account_update)
                                }
                                UpdateOneof::Transaction(tx) => {
                                    summarize_tx(tx)
                                }                    
                                _ => None,
                            }
                        } else {
                            None
                        };

                        if let Some(message) = message {
                            println!("{}", message);
                        }
                    }
                }
            }
        }
    }
}

#[tokio::main]
async fn main() {
    let args: Args = Args::parse();
    let config = std::fs::read_to_string(&args.config).expect("Failed to read config file");
    let config: FumaroleConfig =
        serde_yaml::from_str(&config).expect("Failed to parse config file");

    match args.action {
        Action::Subscribe(sub_args) => {
            subscribe(sub_args, config).await;
        }
    }
}

§High-traffic workload: parallel subscription + zstd

For high-traffic workload or for higher latency connection, using parallel subscription and zstd will greatly improve performance to stay on-tip.

Inside your config.yaml file enable compression with response_compression: zstd:

endpoint: https://fumarole.endpoint.rpcpool.com
x-token: 00000000-0000-0000-0000-000000000000
response_compression: zstd

Uses FumaroleSubscribeConfig to configure the parallel subscription and zstd compression.

let config: FumaroleConfig = serde_yaml::from_reader("<path/to/config.yaml>").expect("failed to parse fumarole config");

let request = SubscribeRequest {
   transactions: HashMap::from([(
       "f1".to_owned(),
       SubscribeRequestFilterTransactions::default(),
   )]),
   ..Default::default()
};


let mut fumarole_client = FumaroleClient::connect(config)
   .await
   .expect("Failed to connect to fumarole");

let subscribe_config = FumaroleSubscribeConfig {
   num_data_plane_tcp_connections: NonZeroU8::new(4).unwrap(), // maximum of 4 TCP connections is allowed
   ..Default::default()
};

let dragonsmouth_session = fumarole_client
   .dragonsmouth_subscribe_with_config(args.name, request, subscribe_config)
   .await
   .expect("Failed to subscribe");
ⓘ
use futures::StreamExt;
use yellowstone_fumarole_client::FumaroleClient;
use yellowstone_fumarole_client::stream::FumaroleEvent;
use yellowstone_grpc_proto::geyser::SubscribeRequest;

async fn run(client: &mut FumaroleClient, subscriber_name: String, request: SubscribeRequest) {
    let subscription = client
        .subscribe(subscriber_name, request)
        .await
        .expect("subscribe failed");

    let (_sink, stream) = subscription.split();
    let mut slot_stream = stream.slot_sequential();

    while let Some(item) = slot_stream.next().await {
        match item.expect("stream error") {
            FumaroleEvent::Data { slot, .. } => {
                println!("data for slot {slot}");
            }
            FumaroleEvent::SlotEnded(slot) => {
                println!("slot ended: {slot}");
            }
        }
    }
}

§BlockStream / Blocstream example (block-oriented processing)

ⓘ
use futures::StreamExt;
use yellowstone_fumarole_client::FumaroleClient;
use yellowstone_fumarole_client::stream::FumaroleBlockStreamEvent;
use yellowstone_grpc_proto::geyser::SubscribeRequest;

async fn run(client: &mut FumaroleClient, subscriber_name: String, request: SubscribeRequest) {
    let subscription = client
        .subscribe(subscriber_name, request)
        .await
        .expect("subscribe failed");

    let (_sink, stream) = subscription.split();
    let mut block_stream = stream.block_stream();

    while let Some(item) = block_stream.next().await {
        match item.expect("stream error") {
            FumaroleBlockStreamEvent::Block(block) => {
                println!("block ready for slot {}", block.slot);
            }
            FumaroleBlockStreamEvent::SlotStatus(status) => {
                println!("slot status update for slot {}", status.slot);
            }
        }
    }
}

§Manual commit example (auto_commit: false)

ⓘ
use futures::StreamExt;
use yellowstone_fumarole_client::{FumaroleClient, FumaroleSubscribeConfig};
use yellowstone_fumarole_client::stream::FumaroleEvent;
use yellowstone_grpc_proto::geyser::SubscribeRequest;

async fn run(client: &mut FumaroleClient, subscriber_name: String, request: SubscribeRequest) {
    let subscribe_config = FumaroleSubscribeConfig {
        auto_commit: false,
        ..Default::default()
    };

    let subscription = client
        .subscribe_with_config(subscriber_name, request, subscribe_config)
        .await
        .expect("subscribe failed");

    let (_sink, stream) = subscription.split();
    let mut slot_stream = stream.slot_sequential();

    while let Some(item) = slot_stream.next().await {
        match item.expect("stream error") {
            FumaroleEvent::Data { .. } => {}
            FumaroleEvent::SlotEnded(slot) => {
                // Commit after your slot processing succeeds.
                slot_stream.commit();
                println!("committed progress at end of slot {slot}");
            }
        }
    }
}

§Enable Prometheus Metrics

To enable Prometheus metrics, add the features = [prometheus] to your Cargo.toml file:

[dependencies]
yellowstone-fumarole-client = { version = "x.y.z", features = ["prometheus"] }

Then, you can use the metrics module to register and expose metrics:

use yellowstone_fumarole_client::metrics;
use prometheus::{Registry};

let r = Registry::new();

metrics::register_metrics(&r);

// After registering, you should see `fumarole_` prefixed metrics in the registry.

§Getting Started

Follows the instruction in the README file to get started.

§Feature Flags

  • prometheus: Enables Prometheus metrics for the Fumarole client.

Re-exports§

pub use crate::error::ConnectError;
pub use crate::error::DataplaneErrorKind;
pub use crate::error::DataplaneStreamError;
pub use crate::error::FumaroleSubscribeError;
pub use crate::error::InvalidMetadataHeader;
pub use crate::stream::DragonsmouthLike;
pub use crate::stream::FumaroleEvent;
pub use crate::stream::FumaroleSink;
pub use crate::stream::FumaroleStream;

Modules§

config
error
proto
stream

Structs§

DragonsmouthAdapterSessionDeprecated
Dragonsmouth flavor fumarole session. Mimics the same API as dragonsmouth but uses fumarole as the backend.
FumaroleClient
Yellowstone Fumarole SDK.
FumaroleClientBuilder
A builder for creating a FumaroleClient.
FumaroleSubscribeConfig
Configuration for the Fumarole subscription session
FumaroleSubscription

Constants§

DEFAULT_COMMIT_INTERVAL
Default Fumarole commit offset interval
DEFAULT_CONCURRENT_DOWNLOAD_LIMIT_PER_TCP
Default maximum number of concurrent download requests to the fumarole service inside a single data plane TCP connection.
DEFAULT_DRAGONSMOUTH_CAPACITY
Default gRPC buffer capacity
DEFAULT_MAX_SLOT_DOWNLOAD_ATTEMPT
Default maximum number of consecutive failed slot download attempts before failing the fumarole session.
DEFAULT_PARA_DATA_STREAMS
Default number of parallel data streams (TCP connections) to open to fumarole.
DEFAULT_REFRESH_TIP_INTERVAL
Default refresh tip interval for the fumarole client. Only useful if you enable prometheus feature flags. grpc_tx