Merge branch 'master' into update-client

CHANGELOG.md

@@ -12,9 +12,16 @@ The minor version will be incremented upon a breaking change and the patch versi
 
 ### Fixes
 
+- deps: update `h2` crate (`RUSTSEC-2024-0332`) ([#316](https://github.com/rpcpool/yellowstone-grpc/pull/316))
+
 ### Features
 
 - client: add gRPC channel options to Node.js ([#306](https://github.com/rpcpool/yellowstone-grpc/pull/306))
+- geyser: add `transactions_status` filter ([#310](https://github.com/rpcpool/yellowstone-grpc/pull/310))
+- geyser: add metric `slot_status_plugin` ([#312](https://github.com/rpcpool/yellowstone-grpc/pull/312))
+- geyser: wrap `geyser_loop` with `unconstrained` ([#313](https://github.com/rpcpool/yellowstone-grpc/pull/313))
+- geyser: handle `/debug_clients` on prometheus endpoint ([#314](https://github.com/rpcpool/yellowstone-grpc/pull/314))
+- geyser: wrap messages to `Arc` ([#315](https://github.com/rpcpool/yellowstone-grpc/pull/315))
 
 ### Breaking
 

examples/rust/src/bin/client.rs

@@ -3,7 +3,7 @@ use {
     clap::{Parser, Subcommand, ValueEnum},
     futures::{future::TryFutureExt, sink::SinkExt, stream::StreamExt},
     log::{error, info},
-    solana_sdk::{pubkey::Pubkey, signature::Signature},
+    solana_sdk::{pubkey::Pubkey, signature::Signature, transaction::TransactionError},
     solana_transaction_status::{EncodedTransactionWithStatusMeta, UiTransactionEncoding},
     std::{collections::HashMap, env, fmt, fs::File, sync::Arc, time::Duration},
     tokio::sync::Mutex,
@@ -17,13 +17,14 @@ use {
         SubscribeRequestFilterBlocks, SubscribeRequestFilterBlocksMeta,
         SubscribeRequestFilterEntry, SubscribeRequestFilterSlots,
         SubscribeRequestFilterTransactions, SubscribeRequestPing, SubscribeUpdateAccount,
-        SubscribeUpdateTransaction,
+        SubscribeUpdateTransaction, SubscribeUpdateTransactionStatus,
     },
 };
 
 type SlotsFilterMap = HashMap<String, SubscribeRequestFilterSlots>;
 type AccountFilterMap = HashMap<String, SubscribeRequestFilterAccounts>;
 type TransactionsFilterMap = HashMap<String, SubscribeRequestFilterTransactions>;
+type TransactionsStatusFilterMap = HashMap<String, SubscribeRequestFilterTransactions>;
 type EntryFilterMap = HashMap<String, SubscribeRequestFilterEntry>;
 type BlocksFilterMap = HashMap<String, SubscribeRequestFilterBlocks>;
 type BlocksMetaFilterMap = HashMap<String, SubscribeRequestFilterBlocksMeta>;
@@ -169,6 +170,34 @@ struct ActionSubscribe {
     #[clap(long)]
     transactions_account_required: Vec<String>,
 
+    /// Subscribe on transactions_status updates
+    #[clap(long)]
+    transactions_status: bool,
+
+    /// Filter vote transactions for transactions_status
+    #[clap(long)]
+    transactions_status_vote: Option<bool>,
+
+    /// Filter failed transactions for transactions_status
+    #[clap(long)]
+    transactions_status_failed: Option<bool>,
+
+    /// Filter by transaction signature for transactions_status
+    #[clap(long)]
+    transactions_status_signature: Option<String>,
+
+    /// Filter included account in transactions for transactions_status
+    #[clap(long)]
+    transactions_status_account_include: Vec<String>,
+
+    /// Filter excluded account in transactions for transactions_status
+    #[clap(long)]
+    transactions_status_account_exclude: Vec<String>,
+
+    /// Filter required account in transactions for transactions_status
+    #[clap(long)]
+    transactions_status_account_required: Vec<String>,
+
     #[clap(long)]
     entry: bool,
 
@@ -289,6 +318,21 @@ impl Action {
                     );
                 }
 
+                let mut transactions_status: TransactionsStatusFilterMap = HashMap::new();
+                if args.transactions_status {
+                    transactions_status.insert(
+                        "client".to_string(),
+                        SubscribeRequestFilterTransactions {
+                            vote: args.transactions_status_vote,
+                            failed: args.transactions_status_failed,
+                            signature: args.transactions_status_signature.clone(),
+                            account_include: args.transactions_status_account_include.clone(),
+                            account_exclude: args.transactions_status_account_exclude.clone(),
+                            account_required: args.transactions_status_account_required.clone(),
+                        },
+                    );
+                }
+
                 let mut entry: EntryFilterMap = HashMap::new();
                 if args.entry {
                     entry.insert("client".to_owned(), SubscribeRequestFilterEntry {});
@@ -333,6 +377,7 @@ impl Action {
                         slots,
                         accounts,
                         transactions,
+                        transactions_status,
                         entry,
                         blocks,
                         blocks_meta,
@@ -429,6 +474,29 @@ impl From<SubscribeUpdateTransaction> for TransactionPretty {
     }
 }
 
+#[allow(dead_code)]
+#[derive(Debug)]
+pub struct TransactionStatusPretty {
+    slot: u64,
+    signature: Signature,
+    is_vote: bool,
+    index: u64,
+    err: Option<TransactionError>,
+}
+
+impl From<SubscribeUpdateTransactionStatus> for TransactionStatusPretty {
+    fn from(status: SubscribeUpdateTransactionStatus) -> Self {
+        Self {
+            slot: status.slot,
+            signature: Signature::try_from(status.signature.as_slice()).expect("valid signature"),
+            is_vote: status.is_vote,
+            index: status.index,
+            err: yellowstone_grpc_proto::convert_from::create_tx_error(status.err.as_ref())
+                .expect("valid tx err"),
+        }
+    }
+}
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     env::set_var(
@@ -557,6 +625,14 @@ async fn geyser_subscribe(
                         );
                         continue;
                     }
+                    Some(UpdateOneof::TransactionStatus(status)) => {
+                        let status: TransactionStatusPretty = status.into();
+                        info!(
+                            "new transaction update: filters {:?}, transaction status: {:?}",
+                            msg.filters, status
+                        );
+                        continue;
+                    }
                     Some(UpdateOneof::Ping(_)) => {
                         // This is necessary to keep load balancers that expect client pings alive. If your load balancer doesn't
                         // require periodic client pings then this is unnecessary
@@ -588,6 +664,7 @@ async fn geyser_subscribe(
                     slots: new_slots.clone(),
                     accounts: HashMap::default(),
                     transactions: HashMap::default(),
+                    transactions_status: HashMap::default(),
                     entry: HashMap::default(),
                     blocks: HashMap::default(),
                     blocks_meta: HashMap::default(),

examples/rust/src/bin/tx-blocktime.rs

@@ -94,7 +94,8 @@ async fn main() -> anyhow::Result<()> {
         .send(SubscribeRequest {
             slots: HashMap::new(),
             accounts: HashMap::new(),
-            transactions: hashmap! { "".to_owned() => SubscribeRequestFilterTransactions {
+            transactions: HashMap::new(),
+            transactions_status: hashmap! { "".to_owned() => SubscribeRequestFilterTransactions {
                 vote: args.vote,
                 failed: args.failed,
                 signature: args.signature,
@@ -116,9 +117,9 @@ async fn main() -> anyhow::Result<()> {
         match message {
             Ok(msg) => {
                 match msg.update_oneof {
-                    Some(UpdateOneof::Transaction(tx)) => {
+                    Some(UpdateOneof::TransactionStatus(tx)) => {
                         let entry = messages.entry(tx.slot).or_default();
-                        let sig = Signature::try_from(tx.transaction.unwrap().signature.as_slice())
+                        let sig = Signature::try_from(tx.signature.as_slice())
                             .expect("valid signature from transaction")
                             .to_string();
                         if let Some(timestamp) = entry.0 {