Stream message

StreamMessageClient wraps the receiving end of a Tokio mpsc channel. Anything you send into the channel as a UnifiedMessage reaches the pipeline as an account or transaction update. Use it to plug in a source Carbon doesn't ship (Kafka, a message queue, a file replay, your own Geyser plugin) or to drive a pipeline from tests, without implementing the Datasource trait yourself.

Quick example#

This program builds three Orca Whirlpool FeeTier accounts by hand, sends them through the channel, and lets the pipeline decode them. It runs offline.

cargo new stream-message && cd stream-message
cargo add carbon-core@2.0.0 carbon-log-metrics@2.0.0 carbon-orca-whirlpool-decoder@2.0.0 carbon-stream-message-datasource@2.0.0
cargo add solana-account@4.3.2 solana-pubkey@4.2.1
cargo add tokio@1 --features full
cargo add env_logger@0.11 log@0.4

use {
    carbon_core::{
        account::AccountProcessorInputType, datasource::AccountUpdate, error::CarbonResult,
        pipeline::Pipeline, processor::Processor,
    },
    carbon_log_metrics::LogMetrics,
    carbon_orca_whirlpool_decoder::{
        accounts::OrcaWhirlpoolAccount, OrcaWhirlpoolDecoder, PROGRAM_ID,
    },
    carbon_stream_message_datasource::{StreamMessageClient, UnifiedMessage},
    solana_account::Account,
    solana_pubkey::Pubkey,
    std::sync::Arc,
};

#[tokio::main]
async fn main() -> CarbonResult<()> {
    env_logger::init();

    let (tx, rx) = tokio::sync::mpsc::channel::<UnifiedMessage>(1_000);
    let datasource = StreamMessageClient::new(rx);

    // Your producer: a Kafka consumer, a file replay, a test fixture...
    tokio::spawn(async move {
        for (slot, tick_spacing) in [(1_u64, 1_u16), (2, 64), (3, 128)] {
            let update = AccountUpdate {
                pubkey: Pubkey::new_unique(),
                account: fee_tier_account(tick_spacing),
                slot,
                transaction_signature: None,
            };
            tx.send(UnifiedMessage::Account(update)).await.unwrap();
        }
        // Dropping `tx` closes the stream; the pipeline exits once it drains.
    });

    Pipeline::builder()
        .datasource(datasource)
        .metrics(Arc::new(LogMetrics::new()))
        .account(OrcaWhirlpoolDecoder, FeeTierLogger)
        .build()?
        .run()
        .await
}

/// Builds the raw bytes of an Orca `FeeTier` account: discriminator, config, tick spacing, fee rate.
fn fee_tier_account(tick_spacing: u16) -> Account {
    let mut data = vec![56, 75, 159, 76, 142, 68, 190, 105];
    data.extend_from_slice(Pubkey::new_unique().as_ref());
    data.extend_from_slice(&tick_spacing.to_le_bytes());
    data.extend_from_slice(&3000_u16.to_le_bytes());
    Account {
        lamports: 1_200_000,
        data,
        owner: PROGRAM_ID,
        executable: false,
        rent_epoch: u64::MAX,
    }
}

struct FeeTierLogger;

impl Processor<AccountProcessorInputType<'_, OrcaWhirlpoolAccount>> for FeeTierLogger {
    async fn process(
        &mut self,
        input: &AccountProcessorInputType<'_, OrcaWhirlpoolAccount>,
    ) -> CarbonResult<()> {
        if let OrcaWhirlpoolAccount::FeeTier(fee_tier) = &input.decoded_account.data {
            println!(
                "slot {} fee tier: tick_spacing={} default_fee_rate={}",
                input.metadata.slot, fee_tier.tick_spacing, fee_tier.default_fee_rate
            );
        }
        Ok(())
    }
}

cargo run

slot 1 fee tier: tick_spacing=1 default_fee_rate=3000
slot 2 fee tier: tick_spacing=64 default_fee_rate=3000
slot 3 fee tier: tick_spacing=128 default_fee_rate=3000

The program exits on its own after the third line.

Configuration#

StreamMessageClient::new#

pub fn new(receiver: tokio::sync::mpsc::Receiver<UnifiedMessage>) -> Self

The channel's capacity is yours to choose; it buffers messages before they reach the pipeline's own channel.

UnifiedMessage#

pub enum UnifiedMessage {
    Account(AccountUpdate),
    Transaction(Box<TransactionUpdate>),
}

AccountUpdate and TransactionUpdate are the types from carbon_core::datasource; see the updates reference for their fields. There is no variant for block details or explicit deletions.

handle_message_stream#

pub async fn handle_message_stream(
    receiver: Receiver<UnifiedMessage>,
    cancellation_token: CancellationToken,
    sender: Sender<(Update, DatasourceId)>,
    id: DatasourceId,
)

The loop StreamMessageClient runs internally, exported for use in your own Datasource implementation.

Behaviour#

• Accounts. Converted with AccountUpdate::into_update(): zero-lamport accounts become AccountDeletion, others AccountUpdate.
• Transactions. Forwarded unchanged, so you set signature, meta, slot, is_vote and the rest yourself. Carbon does not filter failed transactions here.
• Ordering and backpressure. Messages are forwarded in channel order with send().await; nothing is dropped.
• Finishing. When every Sender for your channel is dropped, the loop logs Receiver closed and ends. A pipeline with only this datasource then shuts down.
• Single use. The receiver is taken on the first consume. A second call logs StreamMessageClient.consume() called more than once; receiver already taken and does nothing.
• Metrics. The names start with agave_grpc_: counters agave_grpc_account_updates_received_total, agave_grpc_account_deletions_received_total, agave_grpc_transaction_updates_received_total; histograms agave_grpc_account_process_time_nanoseconds, agave_grpc_account_deletion_process_time_nanoseconds, agave_grpc_transaction_process_time_nanoseconds.

Troubleshooting#

The pipeline never exits#

A Sender clone is still alive somewhere, so the stream never closes. Drop every clone when your producer is done.

The pipeline exits immediately#

Every sender was dropped before anything was sent, or the producer task panicked. Check that the producer holds tx for as long as it sends.

Accounts aren't decoded#

The decoder checks account.owner and the account discriminator. Set owner to the program ID and put the raw on-chain bytes, discriminator included, in data.

The README example doesn't compile#

The crate's README refers to StreamMessageDatasource and pipeline.register; neither exists. The type is StreamMessageClient, added with Pipeline::builder().datasource(…) as above.

See also#

• Custom datasource
• Testing and test utilities
• Updates reference
• Datasources concept