anchor_client/

lib.rs

#![cfg_attr(docsrs, feature(doc_cfg))]

//! An RPC client to interact with Solana programs written in [`anchor_lang`].
//!
//! # Examples
//!
//! A simple example that creates a client, sends a transaction and fetches an account:
//!
//! ```ignore
//! use std::rc::Rc;
//!
//! use anchor_client::{Client, Cluster, Signer};
//! use my_program::{accounts, instruction, MyAccount};
//! use solana_keypair::{read_keypair_file, Keypair};
//! use solana_system_interface::program as system_program;
//!
//! fn main() -> Result<(), Box<dyn std::error::Error>> {
//!     // Create client
//!     let payer = read_keypair_file("keypair.json")?;
//!     let client = Client::new(Cluster::Localnet, Rc::new(payer));
//!
//!     // Create program
//!     let program = client.program(my_program::ID)?;
//!
//!     // Send transaction
//!     let my_account_kp = Keypair::new();
//!     program
//!         .request()
//!         .accounts(accounts::Initialize {
//!             my_account: my_account_kp.pubkey(),
//!             payer: program.payer(),
//!             system_program: system_program::ID,
//!         })
//!         .args(instruction::Initialize { field: 42 })
//!         .signer(&my_account_kp)
//!         .send()?;
//!
//!     // Fetch account
//!     let my_account: MyAccount = program.account(my_account_kp.pubkey())?;
//!     assert_eq!(my_account.field, 42);
//!
//!     Ok(())
//! }
//! ```
//!
//! More examples can be found in [here].
//!
//! [here]: https://github.com/otter-sec/anchor/tree/v1.2.2/client/example/src
//!
//! # Features
//!
//! ## `async`
//!
//! The client is blocking by default. To enable asynchronous client, add `async` feature:
//!
//! ```toml
//! anchor-client = { version = "1.2.2", features = ["async"] }
//! ````
//!
//! ## `mock`
//!
//! This feature allows passing in a custom RPC client when creating program instances, which is
//! useful for mocking RPC responses, e.g. via
//! [`RpcClient::new_mock`](solana_rpc_client::nonblocking::rpc_client::RpcClient::new_mock).

#[cfg(feature = "async")]
pub use nonblocking::ThreadSafeSigner;
use {
    crate::compat::{
        solana_account_decoder, solana_hash, solana_message, solana_pubsub_client,
        solana_rpc_client, solana_rpc_client_api, solana_transaction,
    },
    anchor_lang::{
        solana_program::{program_error::ProgramError, pubkey::Pubkey},
        AccountDeserialize, Discriminator, InstructionData, ToAccountMetas,
    },
    futures::{Future, StreamExt},
    regex::Regex,
    solana_account_decoder::{UiAccount, UiAccountEncoding},
    solana_instruction::AccountMeta,
    solana_message::v0,
    solana_pubsub_client::nonblocking::pubsub_client::PubsubClient,
    solana_rpc_client::nonblocking::rpc_client::RpcClient as AsyncRpcClient,
    solana_rpc_client_api::{
        config::{
            RpcAccountInfoConfig, RpcProgramAccountsConfig, RpcTransactionLogsConfig,
            RpcTransactionLogsFilter,
        },
        filter::Memcmp,
        request::RpcError,
    },
    solana_signature::Signature,
    std::{
        iter::Map,
        marker::PhantomData,
        ops::Deref,
        pin::Pin,
        sync::{Arc, LazyLock},
        vec::IntoIter,
    },
    thiserror::Error,
    tokio::{
        runtime::Handle,
        sync::{
            mpsc::{unbounded_channel, UnboundedReceiver},
            OnceCell,
        },
        task::JoinHandle,
    },
};
pub use {
    anchor_lang,
    cluster::Cluster,
    solana_commitment_config::CommitmentConfig,
    solana_hash::Hash,
    solana_instruction::Instruction,
    solana_message::AddressLookupTableAccount,
    solana_pubsub_client::nonblocking::pubsub_client::PubsubClientError,
    solana_rpc_client_api::{
        client_error::{Error as SolanaClientError, ErrorKind as SolanaClientErrorKind},
        config::RpcSendTransactionConfig,
        filter::RpcFilterType,
    },
    solana_signer::{Signer, SignerError},
    solana_transaction::{versioned::VersionedTransaction, Transaction},
};

mod cluster;
#[doc(hidden)]
pub mod compat;

/// Specifies which transaction version to use when building transactions.
#[derive(Debug, Clone, Default)]
pub enum TxVersion<'a> {
    /// Legacy transaction format.
    #[default]
    Legacy,
    /// Versioned transaction format (v0) with optional address lookup tables.
    V0(&'a [AddressLookupTableAccount]),
}

#[cfg(not(feature = "async"))]
mod blocking;
#[cfg(feature = "async")]
mod nonblocking;

const PROGRAM_LOG: &str = "Program log: ";
const PROGRAM_DATA: &str = "Program data: ";

type UnsubscribeFn = Box<dyn FnOnce() -> Pin<Box<dyn Future<Output = ()> + Send>> + Send>;
/// Client defines the base configuration for building RPC clients to
/// communicate with Anchor programs running on a Solana cluster. It's
/// primary use is to build a `Program` client via the `program` method.
pub struct Client<C> {
    cfg: Config<C>,
}

impl<C: Clone + Deref<Target = impl Signer>> Client<C> {
    pub fn new(cluster: Cluster, payer: C) -> Self {
        Self {
            cfg: Config {
                cluster,
                payer,
                options: None,
            },
        }
    }

    pub fn new_with_options(cluster: Cluster, payer: C, options: CommitmentConfig) -> Self {
        Self {
            cfg: Config {
                cluster,
                payer,
                options: Some(options),
            },
        }
    }

    pub fn program(
        &self,
        program_id: Pubkey,
        #[cfg(feature = "mock")] rpc_client: AsyncRpcClient,
    ) -> Result<Program<C>, ClientError> {
        let cfg = Config {
            cluster: self.cfg.cluster.clone(),
            options: self.cfg.options,
            payer: self.cfg.payer.clone(),
        };

        Program::new(
            program_id,
            cfg,
            #[cfg(feature = "mock")]
            rpc_client,
        )
    }
}

/// Auxiliary data structure to align the types of the Solana CLI utils with Anchor client.
/// Client<C> implementation requires <C: Clone + Deref<Target = impl Signer>> which does not comply with Box<dyn Signer>
/// that's used when loaded Signer from keypair file. This struct is used to wrap the usage.
pub struct DynSigner(pub Arc<dyn Signer>);

impl Signer for DynSigner {
    fn pubkey(&self) -> Pubkey {
        self.0.pubkey()
    }

    fn try_pubkey(&self) -> Result<Pubkey, SignerError> {
        self.0.try_pubkey()
    }

    fn sign_message(&self, message: &[u8]) -> Signature {
        self.0.sign_message(message)
    }

    fn try_sign_message(&self, message: &[u8]) -> Result<Signature, SignerError> {
        self.0.try_sign_message(message)
    }

    fn is_interactive(&self) -> bool {
        self.0.is_interactive()
    }
}

// Internal configuration for a client.
#[derive(Debug)]
pub struct Config<C> {
    cluster: Cluster,
    payer: C,
    options: Option<CommitmentConfig>,
}

pub struct EventUnsubscriber<'a> {
    handle: JoinHandle<Result<(), ClientError>>,
    rx: UnboundedReceiver<UnsubscribeFn>,
    #[cfg(not(feature = "async"))]
    runtime_handle: &'a Handle,
    _lifetime_marker: PhantomData<&'a Handle>,
}

impl EventUnsubscriber<'_> {
    async fn unsubscribe_internal(mut self) {
        if let Some(unsubscribe) = self.rx.recv().await {
            unsubscribe().await;
        }

        let _ = self.handle.await;
    }
}

/// Program is the primary client handle to be used to build and send requests.
pub struct Program<C> {
    program_id: Pubkey,
    cfg: Config<C>,
    sub_client: OnceCell<Arc<PubsubClient>>,
    #[cfg(not(feature = "async"))]
    rt: tokio::runtime::Runtime,
    internal_rpc_client: AsyncRpcClient,
}

impl<C: Deref<Target = impl Signer> + Clone> Program<C> {
    pub fn payer(&self) -> Pubkey {
        self.cfg.payer.pubkey()
    }

    pub fn id(&self) -> Pubkey {
        self.program_id
    }

    #[cfg(feature = "mock")]
    pub fn internal_rpc(&self) -> &AsyncRpcClient {
        &self.internal_rpc_client
    }

    async fn account_internal<T: AccountDeserialize>(
        &self,
        address: Pubkey,
    ) -> Result<T, ClientError> {
        let account = self
            .internal_rpc_client
            .get_account_with_commitment(&address, self.internal_rpc_client.commitment())
            .await
            .map_err(Box::new)?
            .value
            .ok_or(ClientError::AccountNotFound)?;
        let mut data: &[u8] = &account.data;
        T::try_deserialize(&mut data).map_err(Into::into)
    }

    async fn accounts_lazy_internal<T: AccountDeserialize + Discriminator>(
        &self,
        filters: Vec<RpcFilterType>,
    ) -> Result<ProgramAccountsIterator<T>, ClientError> {
        let account_type_filter =
            RpcFilterType::Memcmp(Memcmp::new_base58_encoded(0, T::DISCRIMINATOR));
        let config = RpcProgramAccountsConfig {
            filters: Some([vec![account_type_filter], filters].concat()),
            account_config: RpcAccountInfoConfig {
                encoding: Some(UiAccountEncoding::Base64),
                ..RpcAccountInfoConfig::default()
            },
            ..RpcProgramAccountsConfig::default()
        };

        Ok(ProgramAccountsIterator {
            inner: self
                .internal_rpc_client
                .get_program_ui_accounts_with_config(&self.id(), config)
                .await
                .map_err(Box::new)?
                .into_iter()
                .map(|(key, account)| {
                    let data = account.data.decode().ok_or_else(|| {
                        ClientError::SolanaClientError(Box::new(
                            SolanaClientError::new_with_request(
                                SolanaClientErrorKind::Custom(
                                    "Failed to decode account data".to_string(),
                                ),
                                solana_rpc_client_api::request::RpcRequest::GetProgramAccounts,
                            ),
                        ))
                    })?;
                    Ok((key, T::try_deserialize(&mut data.as_slice())?))
                }),
        })
    }

    async fn on_internal<T: anchor_lang::Event + anchor_lang::AnchorDeserialize>(
        &self,
        mut f: impl FnMut(&EventContext, T) + Send + 'static,
    ) -> Result<
        (
            JoinHandle<Result<(), ClientError>>,
            UnboundedReceiver<UnsubscribeFn>,
        ),
        ClientError,
    > {
        let client = self
            .sub_client
            .get_or_try_init(|| async {
                PubsubClient::new(self.cfg.cluster.ws_url())
                    .await
                    .map(Arc::new)
                    .map_err(|e| ClientError::SolanaClientPubsubError(Box::new(e)))
            })
            .await?
            .clone();

        let (tx, rx) = unbounded_channel::<_>();
        let config = RpcTransactionLogsConfig {
            commitment: self.cfg.options,
        };
        let program_id_str = self.program_id.to_string();
        let filter = RpcTransactionLogsFilter::Mentions(vec![program_id_str.clone()]);

        let handle = tokio::spawn(async move {
            let (mut notifications, unsubscribe) = client
                .logs_subscribe(filter, config)
                .await
                .map_err(Box::new)?;

            tx.send(unsubscribe).map_err(|e| {
                ClientError::SolanaClientPubsubError(Box::new(PubsubClientError::RequestFailed {
                    message: "Unsubscribe failed".to_string(),
                    reason: e.to_string(),
                }))
            })?;

            while let Some(logs) = notifications.next().await {
                if logs.value.err.is_some() {
                    continue;
                }

                let signature: Signature = logs.value.signature.parse().map_err(|e| {
                    ClientError::LogParseError(format!(
                        "Invalid signature '{}': {e}",
                        logs.value.signature
                    ))
                })?;
                let ctx = EventContext {
                    signature,
                    slot: logs.context.slot,
                };
                let events = parse_logs(&logs.value.logs, &program_id_str)?;
                for e in events {
                    f(&ctx, e);
                }
            }
            Ok::<(), ClientError>(())
        });

        Ok((handle, rx))
    }
}

/// Iterator with items of type (Pubkey, T). Used to lazily deserialize account structs.
/// Wrapper type hides the inner type from usages so the implementation can be changed.
pub struct ProgramAccountsIterator<T> {
    inner: Map<IntoIter<(Pubkey, UiAccount)>, AccountConverterFunction<T>>,
}

/// Function type that accepts solana accounts and returns deserialized anchor accounts
type AccountConverterFunction<T> = fn((Pubkey, UiAccount)) -> Result<(Pubkey, T), ClientError>;

impl<T> Iterator for ProgramAccountsIterator<T> {
    type Item = Result<(Pubkey, T), ClientError>;

    fn next(&mut self) -> Option<Self::Item> {
        self.inner.next()
    }
}

pub fn handle_program_log<T: anchor_lang::Event + anchor_lang::AnchorDeserialize>(
    self_program_str: &str,
    l: &str,
) -> Result<(Option<T>, Option<String>, bool), ClientError> {
    use {
        anchor_lang::__private::base64,
        base64::{engine::general_purpose::STANDARD, Engine},
    };

    // Log emitted from the current program.
    if let Some(log) = l
        .strip_prefix(PROGRAM_LOG)
        .or_else(|| l.strip_prefix(PROGRAM_DATA))
    {
        let log_bytes = match STANDARD.decode(log) {
            Ok(log_bytes) => log_bytes,
            _ => {
                #[cfg(feature = "debug")]
                println!("Could not base64 decode log: {}", log);
                return Ok((None, None, false));
            }
        };

        let event = log_bytes
            .starts_with(T::DISCRIMINATOR)
            .then(|| {
                let mut data = &log_bytes[T::DISCRIMINATOR.len()..];
                T::deserialize(&mut data).map_err(|e| ClientError::LogParseError(e.to_string()))
            })
            .transpose()?;

        Ok((event, None, false))
    }
    // System log.
    else {
        let (program, did_pop) = handle_system_log(self_program_str, l);
        Ok((None, program, did_pop))
    }
}

pub fn handle_system_log(this_program_str: &str, log: &str) -> (Option<String>, bool) {
    static INVOKE_RE: LazyLock<Regex> = LazyLock::new(|| {
        Regex::new(r"^Program ([1-9A-HJ-NP-Za-km-z]+) invoke \[([\d]+)\]$").unwrap()
    });
    if let Some(invoke_match) = INVOKE_RE.captures(log) {
        if invoke_match.get(1).unwrap().as_str() == this_program_str {
            return (Some(this_program_str.to_string()), false);

            // `Invoke [1]` instructions are pushed to the stack in `parse_logs`,
            // so this ensures we only push CPIs to the stack at this stage
        } else if invoke_match.get(2).unwrap().as_str() != "1" {
            return (Some("cpi".to_string()), false); // Any string will do.
        }
    }

    if log.starts_with(&format!("Program {this_program_str} log:")) {
        (Some(this_program_str.to_string()), false)
    } else {
        static SUCCESS_RE: LazyLock<Regex> =
            LazyLock::new(|| Regex::new(r"^Program ([1-9A-HJ-NP-Za-km-z]+) success$").unwrap());
        if SUCCESS_RE.is_match(log) {
            (None, true)
        } else {
            (None, false)
        }
    }
}

pub struct Execution {
    stack: Vec<String>,
}

impl Execution {
    pub fn new(logs: &mut &[String]) -> Result<Self, ClientError> {
        let l = &logs[0];
        *logs = &logs[1..];
        static RE: LazyLock<Regex> = LazyLock::new(|| {
            Regex::new(r"^Program ([1-9A-HJ-NP-Za-km-z]+) invoke \[[\d]+\]$").unwrap()
        });
        let c = RE
            .captures(l)
            .ok_or_else(|| ClientError::LogParseError(l.to_string()))?;
        let program = c
            .get(1)
            .ok_or_else(|| ClientError::LogParseError(l.to_string()))?
            .as_str()
            .to_string();
        Ok(Self {
            stack: vec![program],
        })
    }

    /// The program currently on top of the stack.
    ///
    /// # Panics
    ///
    /// Panics if the stack is empty. Prefer [`Execution::try_program`], which
    /// returns `None` instead; the stack legitimately empties whenever a
    /// top-level instruction returns, and more logs can still follow it.
    pub fn program(&self) -> String {
        assert!(!self.stack.is_empty());
        self.stack[self.stack.len() - 1].clone()
    }

    /// The program currently on top of the stack, or `None` when no
    /// instruction is in scope.
    pub fn try_program(&self) -> Option<String> {
        self.stack.last().cloned()
    }

    pub fn push(&mut self, new_program: String) {
        self.stack.push(new_program);
    }

    /// Pops the innermost program off the stack. A no-op when the stack is
    /// already empty, which happens on a `Program <id> success` line that the
    /// runtime emits without a matching tracked `invoke`.
    pub fn pop(&mut self) {
        self.stack.pop();
    }
}

#[derive(Debug)]
pub struct EventContext {
    pub signature: Signature,
    pub slot: u64,
}

#[derive(Debug, Error)]
pub enum ClientError {
    #[error("Account not found")]
    AccountNotFound,
    #[error("{0}")]
    AnchorError(#[from] anchor_lang::error::Error),
    #[error("{0}")]
    ProgramError(#[from] ProgramError),
    #[error("{0}")]
    SolanaClientError(#[from] Box<SolanaClientError>),
    #[error("{0}")]
    SolanaClientPubsubError(#[from] Box<PubsubClientError>),
    #[error("Unable to parse log: {0}")]
    LogParseError(String),
    #[error(transparent)]
    IOError(#[from] std::io::Error),
    #[error("{0}")]
    SignerError(#[from] SignerError),
}

impl ClientError {
    /// Adding a new variant to [`ClientError`] is a breaking change in v1. To mitigate this issue,
    /// use this helper method for all errors that cannot be precisely described by [`ClientError`].
    fn other<E>(e: E) -> Self
    where
        E: Into<Box<dyn std::error::Error + Send + Sync>>,
    {
        Self::IOError(std::io::Error::other(e))
    }
}

pub trait AsSigner {
    fn as_signer(&self) -> &dyn Signer;
}

impl AsSigner for Box<dyn Signer + '_> {
    fn as_signer(&self) -> &dyn Signer {
        self.as_ref()
    }
}

/// `RequestBuilder` provides a builder interface to create and send
/// transactions to a cluster.
pub struct RequestBuilder<'a, C, S: 'a> {
    cluster: String,
    program_id: Pubkey,
    accounts: Vec<AccountMeta>,
    options: CommitmentConfig,
    instructions: Vec<Instruction>,
    payer: C,
    instruction_data: Option<Vec<u8>>,
    signers: Vec<S>,
    #[cfg(not(feature = "async"))]
    handle: &'a Handle,
    internal_rpc_client: &'a AsyncRpcClient,
    _phantom: PhantomData<&'a ()>,
}

// Shared implementation for all RequestBuilders
impl<C: Deref<Target = impl Signer> + Clone, S: AsSigner> RequestBuilder<'_, C, S> {
    #[must_use]
    pub fn payer(mut self, payer: C) -> Self {
        self.payer = payer;
        self
    }

    #[must_use]
    pub fn cluster(mut self, url: &str) -> Self {
        self.cluster = url.to_string();
        self
    }

    #[must_use]
    pub fn instruction(mut self, ix: Instruction) -> Self {
        self.instructions.push(ix);
        self
    }

    #[must_use]
    pub fn program(mut self, program_id: Pubkey) -> Self {
        self.program_id = program_id;
        self
    }

    /// Set the accounts to pass to the instruction.
    ///
    /// `accounts` argument can be:
    ///
    /// - Any type that implements [`ToAccountMetas`] trait
    /// - A vector of [`AccountMeta`]s (for remaining accounts)
    ///
    /// Note that the given accounts are appended to the previous list of accounts instead of
    /// overriding the existing ones (if any).
    ///
    /// # Example
    ///
    /// ```ignore
    /// program
    ///     .request()
    ///     // Regular accounts
    ///     .accounts(accounts::Initialize {
    ///         my_account: my_account_kp.pubkey(),
    ///         payer: program.payer(),
    ///         system_program: system_program::ID,
    ///     })
    ///     // Remaining accounts
    ///     .accounts(vec![AccountMeta {
    ///         pubkey: remaining,
    ///         is_signer: true,
    ///         is_writable: true,
    ///     }])
    ///     .args(instruction::Initialize { field: 42 })
    ///     .send()?;
    /// ```
    #[must_use]
    pub fn accounts(mut self, accounts: impl ToAccountMetas) -> Self {
        let mut metas = accounts.to_account_metas(None);
        self.accounts.append(&mut metas);
        self
    }

    #[must_use]
    pub fn options(mut self, options: CommitmentConfig) -> Self {
        self.options = options;
        self
    }

    #[must_use]
    pub fn args(mut self, args: impl InstructionData) -> Self {
        self.instruction_data = Some(args.data());
        self
    }

    pub fn instructions(&self) -> Vec<Instruction> {
        let mut instructions = self.instructions.clone();
        if let Some(ix_data) = &self.instruction_data {
            instructions.push(Instruction {
                program_id: self.program_id,
                data: ix_data.clone(),
                accounts: self.accounts.clone(),
            });
        }

        instructions
    }

    /// Build the request into a transaction.
    ///
    /// Note: This will build a transaction with the legacy transaction format. If you'd like to use
    /// a different transaction format, use [`transaction_versioned`].
    pub fn transaction(&self) -> Transaction {
        let instructions = &self.instructions();
        Transaction::new_with_payer(instructions, Some(&self.payer.pubkey()))
    }

    /// Build an unsigned transaction.
    ///
    /// # Arguments
    ///
    /// * `version` - The transaction version to use ([`TxVersion::Legacy`] or [`TxVersion::V0`]).
    /// * `recent_blockhash` - A recent blockhash to include in the transaction message.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use anchor_client::{Client, Cluster, TxVersion};
    /// use anchor_lang::prelude::Pubkey;
    /// use solana_signer::null_signer::NullSigner;
    /// use anchor_client::AddressLookupTableAccount;
    /// use anchor_client::Hash;
    ///
    /// let payer = NullSigner::new(&Pubkey::default());
    /// let client = Client::new(Cluster::Localnet, std::rc::Rc::new(payer));
    ///
    /// let program = client.program(Pubkey::default()).unwrap();
    /// // Dummy blockhash
    /// let blockhash = Hash::from([0; 32]);
    /// let lookup_table = AddressLookupTableAccount { key: Pubkey::default(), addresses: vec![] };
    ///
    /// let request = program.request();
    /// // Legacy transaction
    /// let tx = request.transaction_versioned(TxVersion::Legacy, blockhash).unwrap();
    ///
    /// // V0 transaction with address lookup tables
    /// let tx = request.transaction_versioned(TxVersion::V0(&[lookup_table]), blockhash).unwrap();
    ///
    /// // V0 transaction without lookup tables
    /// let tx = request.transaction_versioned(TxVersion::V0(&[]), blockhash).unwrap();
    //// ```
    pub fn transaction_versioned(
        &self,
        version: TxVersion<'_>,
        recent_blockhash: Hash,
    ) -> Result<solana_transaction::versioned::VersionedTransaction, ClientError> {
        let instructions = self.instructions();
        let payer = self.payer.pubkey();

        match version {
            TxVersion::Legacy => {
                let message = solana_message::legacy::Message::new_with_blockhash(
                    &instructions,
                    Some(&payer),
                    &recent_blockhash,
                );
                Ok(solana_transaction::versioned::VersionedTransaction {
                    signatures: vec![
                        solana_signature::Signature::default();
                        message.header.num_required_signatures as usize
                    ],
                    message: solana_message::VersionedMessage::Legacy(message),
                })
            }
            TxVersion::V0(address_lookup_table_accounts) => {
                let message = v0::Message::try_compile(
                    &payer,
                    &instructions,
                    address_lookup_table_accounts,
                    recent_blockhash,
                )
                .map_err(ClientError::other)?;
                Ok(solana_transaction::versioned::VersionedTransaction {
                    signatures: vec![
                        solana_signature::Signature::default();
                        message.header.num_required_signatures as usize
                    ],
                    message: solana_message::VersionedMessage::V0(message),
                })
            }
        }
    }

    fn signed_transaction_with_blockhash_versioned(
        &self,
        version: TxVersion<'_>,
        latest_hash: Hash,
    ) -> Result<solana_transaction::versioned::VersionedTransaction, ClientError> {
        let signers: Vec<&dyn Signer> = self.signers.iter().map(|s| s.as_signer()).collect();
        let mut all_signers = signers;
        all_signers.push(&*self.payer);

        let instructions = self.instructions();
        let payer = self.payer.pubkey();

        let message = match version {
            TxVersion::Legacy => {
                let msg = solana_message::legacy::Message::new_with_blockhash(
                    &instructions,
                    Some(&payer),
                    &latest_hash,
                );
                solana_message::VersionedMessage::Legacy(msg)
            }
            TxVersion::V0(address_lookup_table_accounts) => {
                let msg = v0::Message::try_compile(
                    &payer,
                    &instructions,
                    address_lookup_table_accounts,
                    latest_hash,
                )
                .map_err(ClientError::other)?;
                solana_message::VersionedMessage::V0(msg)
            }
        };

        let tx =
            solana_transaction::versioned::VersionedTransaction::try_new(message, &all_signers)?;

        Ok(tx)
    }

    async fn signed_transaction_internal(
        &self,
        version: TxVersion<'_>,
    ) -> Result<solana_transaction::versioned::VersionedTransaction, ClientError> {
        let latest_hash = self
            .internal_rpc_client
            .get_latest_blockhash_with_commitment(self.options)
            .await
            .map_err(Box::new)?
            .0;

        self.signed_transaction_with_blockhash_versioned(version, latest_hash)
    }

    async fn send_internal(&self, version: TxVersion<'_>) -> Result<Signature, ClientError> {
        let (latest_hash, _) = self
            .internal_rpc_client
            .get_latest_blockhash_with_commitment(self.options)
            .await
            .map_err(Box::new)?;
        let tx = self.signed_transaction_with_blockhash_versioned(version, latest_hash)?;

        // FIXME: Inline a no-spinner version of `RpcClient::send_and_confirm_transaction`
        // that honors the configured commitment level (`self.options`). The built-in
        // non-spinner methods ignore the commitment, and the only commitment-aware
        // confirmation helper (`send_and_confirm_transaction_with_spinner_and_commitment`)
        // forces a spinner onto callers. Replace this with the non-spinner,
        // commitment-aware method once we upgrade to Solana 4.0, which adds it.
        let signature = self
            .internal_rpc_client
            .send_transaction(&tx)
            .await
            .map_err(Box::new)?;

        loop {
            match self
                .internal_rpc_client
                .get_signature_status_with_commitment(&signature, self.options)
                .await
                .map_err(Box::new)?
            {
                Some(Ok(())) => return Ok(signature),
                Some(Err(e)) => return Err(ClientError::SolanaClientError(Box::new(e.into()))),
                None => {
                    if !self
                        .internal_rpc_client
                        .is_blockhash_valid(&latest_hash, CommitmentConfig::processed())
                        .await
                        .map_err(Box::new)?
                    {
                        // Block hash is not found by some reason
                        break;
                    } else if cfg!(not(test)) {
                        // Retry twice a second
                        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
                    }
                }
            }
        }

        Err(ClientError::SolanaClientError(Box::new(
            RpcError::ForUser(
                "unable to confirm transaction. This can happen in situations such as transaction \
                 expiration and insufficient fee-payer funds"
                    .to_string(),
            )
            .into(),
        )))
    }

    async fn send_with_spinner_and_config_internal(
        &self,
        version: TxVersion<'_>,
        config: RpcSendTransactionConfig,
    ) -> Result<Signature, ClientError> {
        let (latest_hash, _) = self
            .internal_rpc_client
            .get_latest_blockhash_with_commitment(self.options)
            .await
            .map_err(Box::new)?;
        let tx = self.signed_transaction_with_blockhash_versioned(version, latest_hash)?;

        self.internal_rpc_client
            .send_and_confirm_transaction_with_spinner_and_config(&tx, self.options, config)
            .await
            .map_err(|e| Box::new(e).into())
    }
}

fn parse_logs<T: anchor_lang::Event + anchor_lang::AnchorDeserialize>(
    logs: &[String],
    program_id_str: &str,
) -> Result<Vec<T>, ClientError> {
    let mut logs = logs;
    let mut events: Vec<T> = Vec::new();
    if !logs.is_empty() {
        if let Ok(mut execution) = Execution::new(&mut logs) {
            // Create a new peekable iterator so that we can peek at the next log whilst iterating
            let mut logs_iter = logs.iter().peekable();
            static RE: LazyLock<Regex> = LazyLock::new(|| {
                Regex::new(r"^Program ([1-9A-HJ-NP-Za-km-z]+) invoke \[(\d+)\]$").unwrap()
            });

            while let Some(l) = logs_iter.next() {
                // No instruction is in scope. This is reached whenever a
                // top-level instruction has returned but the log stream has
                // not ended -- most commonly the runtime's trailing
                // `"Log truncated"` marker, which is appended after the final
                // `success` when a transaction overruns the log buffer.
                //
                // Only a new top-level `invoke [1]` can re-enter a program
                // context; anything else carries no events, so skip it rather
                // than panicking in `Execution::program`.
                let Some(current_program) = execution.try_program() else {
                    if let Some(caps) = RE.captures(l) {
                        if &caps[2] == "1" {
                            execution.push(caps[1].to_string());
                        }
                    }
                    continue;
                };

                // Parse the log.
                let (event, new_program, did_pop) = {
                    if program_id_str == current_program {
                        handle_program_log(program_id_str, l)?
                    } else {
                        let (program, did_pop) = handle_system_log(program_id_str, l);
                        (None, program, did_pop)
                    }
                };
                // Emit the event.
                if let Some(e) = event {
                    events.push(e);
                }
                // Switch program context on CPI.
                if let Some(new_program) = new_program {
                    execution.push(new_program);
                }
                // Program returned.
                if did_pop {
                    execution.pop();

                    // If the current iteration popped then it means there was a
                    //`Program x success` log. If the next log in the iteration is
                    // of depth [1] then we're not within a CPI and this is a new instruction.
                    //
                    // We need to ensure that the `Execution` instance is updated with
                    // the next program ID, or else `execution.program()` will cause
                    // a panic during the next iteration.
                    //
                    // Use the full regex match to gate this branch. A loose
                    // `ends_with("invoke [1]")` check would also accept program-emitted
                    // log lines that happen to end in that suffix (e.g.
                    // `"Program log: ...invoke [1]"`), which then fail the strict
                    // `^Program <pubkey> invoke [N]$` regex and panic on unwrap.
                    if let Some(&next_log) = logs_iter.peek() {
                        if let Some(caps) = RE.captures(next_log) {
                            if &caps[2] == "1" {
                                execution.push(caps[1].to_string());
                            }
                        }
                    };
                }
            }
        }
    }
    Ok(events)
}

#[cfg(test)]
mod tests {
    // Creating a mock struct that implements `anchor_lang::events`
    // for type inference in `test_logs`
    use {
        anchor_lang::{prelude::*, Event},
        futures::{SinkExt, StreamExt},
        std::sync::atomic::{AtomicU64, Ordering},
        tokio_tungstenite::tungstenite::Message,
    };
    #[derive(Debug, Clone, Copy)]
    #[event]
    pub struct MockEvent {}

    use super::*;
    #[test]
    fn new_execution() {
        let mut logs: &[String] =
            &["Program 7Y8VDzehoewALqJfyxZYMgYCnMTCDhWuGfJKUvjYWATw invoke [1]".to_string()];
        let exe = Execution::new(&mut logs).unwrap();
        assert_eq!(
            exe.stack[0],
            "7Y8VDzehoewALqJfyxZYMgYCnMTCDhWuGfJKUvjYWATw".to_string()
        );
    }

    #[test]
    fn handle_system_log_pop() {
        let log = "Program 7Y8VDzehoewALqJfyxZYMgYCnMTCDhWuGfJKUvjYWATw success";
        let (program, did_pop) = handle_system_log("asdf", log);
        assert_eq!(program, None);
        assert!(did_pop);
    }

    #[test]
    fn handle_system_log_no_pop() {
        let log = "Program 7swsTUiQ6KUK4uFYquQKg4epFRsBnvbrTf2fZQCa2sTJ qwer";
        let (program, did_pop) = handle_system_log("asdf", log);
        assert_eq!(program, None);
        assert!(!did_pop);
    }

    #[test]
    fn test_parse_logs() -> Result<()> {
        // These logs are based on a Jupiter transaction.
        let logs = vec![
            "Program VeryCoolProgram invoke [1]", // Outer instruction #1 starts
            "Program log: Instruction: VeryCoolEvent",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA invoke [2]",
            "Program log: Instruction: Transfer",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA consumed 4645 of 664387 compute \
             units",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA success",
            "Program VeryCoolProgram consumed 42417 of 700000 compute units",
            "Program VeryCoolProgram success", // Outer instruction #1 ends
            "Program EvenCoolerProgram invoke [1]", // Outer instruction #2 starts
            "Program log: Instruction: EvenCoolerEvent",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA invoke [2]",
            "Program log: Instruction: TransferChecked",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA consumed 6200 of 630919 compute \
             units",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA success",
            "Program HyaB3W9q6XdA5xwpU4XnSZV94htfmbmqJXZcEbRaJutt invoke [2]",
            "Program log: Instruction: Swap",
            "Program log: INVARIANT: SWAP",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA invoke [3]",
            "Program log: Instruction: Transfer",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA consumed 4736 of 539321 compute \
             units",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA success",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA invoke [3]",
            "Program log: Instruction: Transfer",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA consumed 4645 of 531933 compute \
             units",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA success",
            "Program HyaB3W9q6XdA5xwpU4XnSZV94htfmbmqJXZcEbRaJutt consumed 84670 of 610768 \
             compute units",
            "Program HyaB3W9q6XdA5xwpU4XnSZV94htfmbmqJXZcEbRaJutt success",
            "Program EvenCoolerProgram invoke [2]",
            "Program EvenCoolerProgram consumed 2021 of 523272 compute units",
            "Program EvenCoolerProgram success",
            "Program HyaB3W9q6XdA5xwpU4XnSZV94htfmbmqJXZcEbRaJutt invoke [2]",
            "Program log: Instruction: Swap",
            "Program log: INVARIANT: SWAP",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA invoke [3]",
            "Program log: Instruction: Transfer",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA consumed 4736 of 418618 compute \
             units",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA success",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA invoke [3]",
            "Program log: Instruction: Transfer",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA consumed 4645 of 411230 compute \
             units",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA success",
            "Program HyaB3W9q6XdA5xwpU4XnSZV94htfmbmqJXZcEbRaJutt consumed 102212 of 507607 \
             compute units",
            "Program HyaB3W9q6XdA5xwpU4XnSZV94htfmbmqJXZcEbRaJutt success",
            "Program EvenCoolerProgram invoke [2]",
            "Program EvenCoolerProgram consumed 2021 of 402569 compute units",
            "Program EvenCoolerProgram success",
            "Program 9W959DqEETiGZocYWCQPaJ6sBmUzgfxXfqGeTEdp3aQP invoke [2]",
            "Program log: Instruction: Swap",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA invoke [3]",
            "Program log: Instruction: Transfer",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA consumed 4736 of 371140 compute \
             units",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA success",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA invoke [3]",
            "Program log: Instruction: MintTo",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA consumed 4492 of 341800 compute \
             units",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA success",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA invoke [3]",
            "Program log: Instruction: Transfer",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA consumed 4645 of 334370 compute \
             units",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA success",
            "Program 9W959DqEETiGZocYWCQPaJ6sBmUzgfxXfqGeTEdp3aQP consumed 57610 of 386812 \
             compute units",
            "Program 9W959DqEETiGZocYWCQPaJ6sBmUzgfxXfqGeTEdp3aQP success",
            "Program EvenCoolerProgram invoke [2]",
            "Program EvenCoolerProgram consumed 2021 of 326438 compute units",
            "Program EvenCoolerProgram success",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA invoke [2]",
            "Program log: Instruction: TransferChecked",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA consumed 6173 of 319725 compute \
             units",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA success",
            "Program EvenCoolerProgram consumed 345969 of 657583 compute units",
            "Program EvenCoolerProgram success", // Outer instruction #2 ends
            "Program ComputeBudget111111111111111111111111111111 invoke [1]",
            "Program ComputeBudget111111111111111111111111111111 success",
            "Program ComputeBudget111111111111111111111111111111 invoke [1]",
            "Program ComputeBudget111111111111111111111111111111 success",
        ];

        // Convert to the owned strings received from RPC.
        let logs: Vec<String> = logs.iter().map(|&l| l.to_string()).collect();

        let program_id_str = "VeryCoolProgram";

        // No events returned here. Just ensuring that the function doesn't panic
        // due an incorrectly emptied stack.
        parse_logs::<MockEvent>(&logs, program_id_str).unwrap();

        Ok(())
    }

    #[test]
    fn failed_transaction_events_are_not_emitted() {
        use {
            anchor_lang::__private::base64,
            base64::{engine::general_purpose::STANDARD, Engine},
            std::{sync::mpsc, time::Duration},
        };

        let rt = tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .build()
            .unwrap();
        let (addr_tx, addr_rx) = mpsc::channel();
        let program_id = Pubkey::new_unique();
        let program_id_str = program_id.to_string();
        let event_log = format!("Program data: {}", STANDARD.encode(MockEvent {}.data()));
        let failed_signature = Signature::default();
        let successful_signature = Signature::from([1u8; 64]);

        rt.spawn(async move {
            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            addr_tx.send(listener.local_addr().unwrap()).unwrap();

            let (stream, _) = listener.accept().await.unwrap();
            let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
            ws.next().await.unwrap().unwrap();
            ws.send(Message::Text(
                r#"{"jsonrpc":"2.0","result":0,"id":1}"#.into(),
            ))
            .await
            .unwrap();

            let failed_notification = format!(
                r#"{{"jsonrpc":"2.0","method":"logsNotification","params":{{"result":{{"context":{{"slot":1}},"value":{{"signature":"{failed_signature}","err":"AccountNotFound","logs":["Program {program_id_str} invoke [1]","{event_log}","Program {program_id_str} failed: custom program error: 0x1"]}}}},"subscription":0}}}}"#
            );
            ws.send(Message::Text(failed_notification.into()))
                .await
                .unwrap();

            let successful_notification = format!(
                r#"{{"jsonrpc":"2.0","method":"logsNotification","params":{{"result":{{"context":{{"slot":2}},"value":{{"signature":"{successful_signature}","err":null,"logs":["Program {program_id_str} invoke [1]","{event_log}","Program {program_id_str} success"]}}}},"subscription":0}}}}"#
            );
            ws.send(Message::Text(successful_notification.into()))
                .await
                .unwrap();

            // Keep the websocket alive until the client or test runtime closes it.
            ws.next().await;
        });

        let addr = addr_rx.recv().unwrap();
        let ws_url = format!("ws://{addr}");
        let client = super::Client::new(
            super::Cluster::Custom(ws_url.clone(), ws_url),
            std::sync::Arc::new(solana_keypair::Keypair::new()),
        );
        let program = client.program(program_id).unwrap();
        let (event_tx, event_rx) = mpsc::channel();
        let (handle, _unsubscribe_rx) = rt
            .block_on(program.on_internal(move |ctx, _event: MockEvent| {
                event_tx.send(ctx.signature).unwrap();
            }))
            .unwrap();

        let emitted_signature = event_rx
            .recv_timeout(Duration::from_secs(5))
            .unwrap_or_else(|e| {
                panic!(
                    "event subscription ended without an event ({e}): {:?}",
                    rt.block_on(handle)
                )
            });
        assert_eq!(emitted_signature, successful_signature);
        assert!(event_rx.recv_timeout(Duration::from_millis(100)).is_err());
    }

    #[test]
    fn test_parse_logs_fake_pop() -> Result<()> {
        let logs = [
            "Program fake111111111111111111111111111111111111112 invoke [1]",
            "Program log: i logged success",
            "Program log: i logged success",
            "Program fake111111111111111111111111111111111111112 consumed 1411 of 200000 compute \
             units",
            "Program fake111111111111111111111111111111111111112 success",
        ];

        // Convert to the owned strings received from RPC.
        let logs: Vec<String> = logs.iter().map(|&l| l.to_string()).collect();

        let program_id_str = "TokenzQdBNbLqP5VEhdkAS6EPFLC1PHnBqCXEpPxuEb";

        // No events returned here. Just ensuring that the function doesn't panic
        // due an incorrectly emptied stack.
        parse_logs::<MockEvent>(&logs, program_id_str).unwrap();

        Ok(())
    }

    /// Regression for #4461: a program-emitted `Program log:` line that ends
    /// with the literal `"invoke [1]"` (e.g. log content that happens to
    /// describe a CPI) used to satisfy the `ends_with` gate but fail the
    /// strict `^Program <pubkey> invoke [N]$` regex, panicking on
    /// `.captures(...).unwrap()` whenever it appeared right after a CPI pop.
    #[test]
    fn test_parse_logs_log_line_ends_with_invoke_1() -> Result<()> {
        let logs = [
            "Program VeryCoolProgram invoke [1]",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA invoke [2]",
            "Program TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA success",
            // Program-emitted log that happens to end with "invoke [1]"
            // immediately after a CPI returns. Pre-fix this would panic.
            "Program log: forwarded inner instruction invoke [1]",
            "Program VeryCoolProgram success",
        ];
        let logs: Vec<String> = logs.iter().map(|&l| l.to_string()).collect();

        parse_logs::<MockEvent>(&logs, "VeryCoolProgram").unwrap();

        Ok(())
    }

    #[test]
    fn execution_pop_past_empty_is_not_a_panic() {
        let mut logs: &[String] =
            &["Program term9YPb9mzAsABaqN71A4xdbxHmpBNZavpBiQKZzN3 invoke [1]".to_string()];
        let mut exe = Execution::new(&mut logs).unwrap();
        assert_eq!(
            exe.try_program().as_deref(),
            Some("term9YPb9mzAsABaqN71A4xdbxHmpBNZavpBiQKZzN3")
        );

        exe.pop();
        assert_eq!(exe.try_program(), None);

        // A second pop with nothing left used to trip `assert!(!self.stack.is_empty())`.
        exe.pop();
        assert_eq!(exe.try_program(), None);
    }

    /// Regression for #1941: the runtime appends a bare `"Log truncated"` line
    /// after the final `success` when a transaction overruns the log buffer.
    /// That line arrives with an empty stack, and `Execution::program` used to
    /// panic on it -- taking down the whole `logs_subscribe` thread rather than
    /// returning an error.
    #[test]
    fn test_parse_logs_trailing_log_after_last_instruction() -> Result<()> {
        let logs = [
            "Program ComputeBudget111111111111111111111111111111 invoke [1]",
            "Program ComputeBudget111111111111111111111111111111 success",
            "Log truncated",
        ];
        let logs: Vec<String> = logs.iter().map(|&l| l.to_string()).collect();

        let events =
            parse_logs::<MockEvent>(&logs, "term9YPb9mzAsABaqN71A4xdbxHmpBNZavpBiQKZzN3").unwrap();

        assert!(events.is_empty());

        Ok(())
    }

    /// The empty-stack guard must skip only the logs that carry no events -- a
    /// later top-level `invoke [1]` still has to re-enter the program context,
    /// or the fix would trade a panic for silently dropped events.
    #[test]
    fn test_parse_logs_event_after_trailing_log() -> Result<()> {
        use {
            anchor_lang::__private::base64,
            base64::{engine::general_purpose::STANDARD, Engine},
        };

        let program_data_log = format!("Program data: {}", STANDARD.encode(MockEvent {}.data()));

        let logs = vec![
            "Program ComputeBudget111111111111111111111111111111 invoke [1]".to_string(),
            "Program ComputeBudget111111111111111111111111111111 success".to_string(),
            // Empty stack from here until the next top-level invoke.
            "Log truncated".to_string(),
            "Program term9YPb9mzAsABaqN71A4xdbxHmpBNZavpBiQKZzN3 invoke [1]".to_string(),
            program_data_log,
            "Program term9YPb9mzAsABaqN71A4xdbxHmpBNZavpBiQKZzN3 success".to_string(),
        ];

        let events =
            parse_logs::<MockEvent>(&logs, "term9YPb9mzAsABaqN71A4xdbxHmpBNZavpBiQKZzN3").unwrap();

        assert_eq!(events.len(), 1);

        Ok(())
    }

    #[test]
    fn test_parse_log_response_inner_events() -> Result<()> {
        use {
            anchor_lang::__private::base64,
            base64::{engine::general_purpose::STANDARD, Engine},
        };

        let mock_event = MockEvent {};
        let program_data_log = format!("Program data: {}", STANDARD.encode(mock_event.data()));

        let logs = vec![
            "Program ComputeBudget111111111111111111111111111111 invoke [1]",
            "Program ComputeBudget111111111111111111111111111111 success",
            "Program ComputeBudget111111111111111111111111111111 invoke [1]",
            "Program ComputeBudget111111111111111111111111111111 success",
            "Program term9YPb9mzAsABaqN71A4xdbxHmpBNZavpBiQKZzN3 invoke [1]",
            "Program log: Instruction: ValidateNonce",
            "Program term9YPb9mzAsABaqN71A4xdbxHmpBNZavpBiQKZzN3 consumed 4839 of 239700 compute \
             units",
            "Program term9YPb9mzAsABaqN71A4xdbxHmpBNZavpBiQKZzN3 success",
            "Program term9YPb9mzAsABaqN71A4xdbxHmpBNZavpBiQKZzN3 invoke [1]",
            "Program log: Instruction: SellExactInPumpFunV3",
            "Program 6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P invoke [2]",
            "Program log: Instruction: Sell",
            "Program pfeeUxB6jkeY1Hxd7CsFCAjcbHA9rWtchMGdZ6VojVZ invoke [3]",
            "Program log: Instruction: GetFees",
            "Program pfeeUxB6jkeY1Hxd7CsFCAjcbHA9rWtchMGdZ6VojVZ consumed 3136 of 187774 compute \
             units",
            "Program return: pfeeUxB6jkeY1Hxd7CsFCAjcbHA9rWtchMGdZ6VojVZ \
             AAAAAAAAAABfAAAAAAAAAB4AAAAAAAAA",
            "Program pfeeUxB6jkeY1Hxd7CsFCAjcbHA9rWtchMGdZ6VojVZ success",
            "Program TokenzQdBNbLqP5VEhdkAS6EPFLC1PHnBqCXEpPxuEb invoke [3]",
            "Program log: Instruction: TransferChecked",
            "Program TokenzQdBNbLqP5VEhdkAS6EPFLC1PHnBqCXEpPxuEb consumed 2475 of 180928 compute \
             units",
            "Program TokenzQdBNbLqP5VEhdkAS6EPFLC1PHnBqCXEpPxuEb success",
            &program_data_log,
            "Program 6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P invoke [3]",
            "Program 6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P consumed 2060 of 166037 compute \
             units",
            "Program 6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P success",
            "Program 6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P consumed 60634 of 223605 compute \
             units",
            "Program 6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P success",
            "Program term9YPb9mzAsABaqN71A4xdbxHmpBNZavpBiQKZzN3 consumed 72662 of 234861 compute \
             units",
            "Program term9YPb9mzAsABaqN71A4xdbxHmpBNZavpBiQKZzN3 success",
            "Program 11111111111111111111111111111111 invoke [1]",
            "Program 11111111111111111111111111111111 success",
            "Program 11111111111111111111111111111111 invoke [1]",
            "Program 11111111111111111111111111111111 success",
        ];

        // Convert to the owned strings received from RPC.
        let logs: Vec<String> = logs.iter().map(|&l| l.to_string()).collect();

        let program_id_str = "6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P";

        let events = parse_logs::<MockEvent>(&logs, program_id_str).unwrap();

        assert_eq!(events.len(), 1);

        Ok(())
    }

    /// Regression test that registering multiple event listeners does not deadlock.
    #[test]
    fn multiple_listeners_no_deadlock() {
        // Spin up a tiny mock websocket server that responds to `logsSubscribe`
        // JSON-RPC requests with a valid subscription id.
        let rt = tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .build()
            .unwrap();

        let (addr_tx, addr_rx) = std::sync::mpsc::channel();

        rt.spawn(async move {
            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = listener.local_addr().unwrap();
            addr_tx.send(addr).unwrap();

            static SUB_ID: AtomicU64 = AtomicU64::new(0);

            loop {
                let (stream, _) = listener.accept().await.unwrap();
                tokio::spawn(async move {
                    let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
                    while let Some(Ok(Message::Text(_))) = ws.next().await {
                        let sub_id = SUB_ID.fetch_add(1, Ordering::Relaxed);
                        // The PubsubClient sends sequential integer ids starting at 0.
                        let resp =
                            format!(r#"{{"jsonrpc":"2.0","result":{sub_id},"id":{sub_id}}}"#);
                        ws.send(Message::Text(resp.into())).await.unwrap();
                    }
                });
            }
        });

        let addr = addr_rx.recv().unwrap();
        let ws_url = format!("ws://{}", addr);

        let client = super::Client::new(
            super::Cluster::Custom(ws_url.clone(), ws_url),
            std::sync::Arc::new(solana_keypair::Keypair::new()),
        );
        let program = client.program(Pubkey::new_unique()).unwrap();

        // With the old RwLock-based code, the second call would deadlock.
        // Use a timeout to ensure the test fails instead of hanging forever.
        let (done_tx, done_rx) = std::sync::mpsc::channel();
        let handle = std::thread::spawn(move || {
            #[cfg(not(feature = "async"))]
            {
                let _listener1 = program
                    .on::<MockEvent>(|_ctx, _event| {})
                    .expect("first listener");

                let _listener2 = program
                    .on::<MockEvent>(|_ctx, _event| {})
                    .expect("second listener");
            }

            #[cfg(feature = "async")]
            {
                let rt = tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                    .unwrap();
                rt.block_on(async {
                    let _listener1 = program
                        .on::<MockEvent>(|_ctx, _event| {})
                        .await
                        .expect("first listener");

                    let _listener2 = program
                        .on::<MockEvent>(|_ctx, _event| {})
                        .await
                        .expect("second listener");
                });
            }

            let _ = done_tx.send(());
        });

        // If this times out, the deadlock is still present.
        done_rx
            .recv_timeout(std::time::Duration::from_secs(5))
            .expect("registering two listeners should not deadlock");

        handle.join().unwrap();
    }
}