blockworks-foundation
/
yellowstone-grpc
mirror of https://github.com/blockworks-foundation/yellowstone-grpc.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

geyser: add `transactions_status` filter (#310)

This commit is contained in:
Kirill Fomichev 2024-04-02 20:46:43 -04:00
parent 834b986d47
commit f8d2d30382
No known key found for this signature in database
GPG Key ID: 6AA0144D5E0C0C0A
17 changed files with 3543 additions and 1522 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGELOG.md
View File

@ -15,6 +15,7 @@ The minor version will be incremented upon a breaking change and the patch versi
### 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))
### Breaking

1342
examples/golang/proto/geyser.pb.go
View File

File diff suppressed because it is too large Load Diff

81
examples/rust/src/bin/client.rs
View File

@ -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,
@ -23,7 +23,7 @@ use {
SubscribeRequestFilterBlocks, SubscribeRequestFilterBlocksMeta,
SubscribeRequestFilterEntry, SubscribeRequestFilterSlots,
SubscribeRequestFilterTransactions, SubscribeRequestPing, SubscribeUpdateAccount,
SubscribeUpdateTransaction,
SubscribeUpdateTransaction, SubscribeUpdateTransactionStatus,
},
tonic::service::Interceptor,
},
@ -32,6 +32,7 @@ use {
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>;
@ -167,6 +168,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,
@ -287,6 +316,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 {});
@ -331,6 +375,7 @@ impl Action {
slots,
accounts,
transactions,
transactions_status,
entry,
blocks,
blocks_meta,
@ -427,6 +472,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(
@ -563,6 +631,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
@ -594,6 +670,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(),

7
examples/rust/src/bin/tx-blocktime.rs
View File

@ -91,7 +91,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,
@ -113,9 +114,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 {

2459
yellowstone-grpc-client-nodejs/src/grpc/geyser.ts
View File

File diff suppressed because it is too large Load Diff

850
yellowstone-grpc-client-nodejs/src/grpc/solana-storage.ts
View File

File diff suppressed because it is too large Load Diff

2
yellowstone-grpc-client-nodejs/src/index.ts
View File

@ -92,6 +92,7 @@ export default class Client {
accounts: { [key: string]: SubscribeRequestFilterAccounts },
slots: { [key: string]: SubscribeRequestFilterSlots },
transactions: { [key: string]: SubscribeRequestFilterTransactions },
transactionsStatus: { [key: string]: SubscribeRequestFilterTransactions },
entry: { [key: string]: SubscribeRequestFilterEntry },
blocks: { [key: string]: SubscribeRequestFilterBlocks },
blocksMeta: { [key: string]: SubscribeRequestFilterBlocksMeta },
@ -106,6 +107,7 @@ export default class Client {
accounts,
slots,
transactions,
transactionsStatus,
entry,
blocks,
blocksMeta,

2
yellowstone-grpc-client/src/lib.rs
View File

@ -212,6 +212,7 @@ impl<F: Interceptor> GeyserGrpcClient<F> {
slots: HashMap<String, SubscribeRequestFilterSlots>,
accounts: HashMap<String, SubscribeRequestFilterAccounts>,
transactions: HashMap<String, SubscribeRequestFilterTransactions>,
transactions_status: HashMap<String, SubscribeRequestFilterTransactions>,
entry: HashMap<String, SubscribeRequestFilterEntry>,
blocks: HashMap<String, SubscribeRequestFilterBlocks>,
blocks_meta: HashMap<String, SubscribeRequestFilterBlocksMeta>,
@ -223,6 +224,7 @@ impl<F: Interceptor> GeyserGrpcClient<F> {
slots,
accounts,
transactions,
transactions_status,
entry,
blocks,
blocks_meta,

8
yellowstone-grpc-geyser/config.json
View File

@ -35,6 +35,14 @@
"account_exclude_max": 10,
"account_required_max": 10
},
"transactions_status": {
"max": 1,
"any": false,
"account_include_max": 10,
"account_include_reject": ["TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"],
"account_exclude_max": 10,
"account_required_max": 10
},
"blocks": {
"max": 1,
"account_include_max": 10,

1
yellowstone-grpc-geyser/src/config.rs
View File

@ -138,6 +138,7 @@ pub struct ConfigGrpcFilters {
pub accounts: ConfigGrpcFiltersAccounts,
pub slots: ConfigGrpcFiltersSlots,
pub transactions: ConfigGrpcFiltersTransactions,
pub transactions_status: ConfigGrpcFiltersTransactions,
pub blocks: ConfigGrpcFiltersBlocks,
pub blocks_meta: ConfigGrpcFiltersBlocksMeta,
pub entry: ConfigGrpcFiltersEntry,

275
yellowstone-grpc-geyser/src/filters.rs
View File

@ -33,6 +33,7 @@ pub struct Filter {
accounts: FilterAccounts,
slots: FilterSlots,
transactions: FilterTransactions,
transactions_status: FilterTransactions,
entry: FilterEntry,
blocks: FilterBlocks,
blocks_meta: FilterBlocksMeta,
@ -46,7 +47,16 @@ impl Filter {
Ok(Self {
accounts: FilterAccounts::new(&config.accounts, &limit.accounts)?,
slots: FilterSlots::new(&config.slots, &limit.slots)?,
transactions: FilterTransactions::new(&config.transactions, &limit.transactions)?,
transactions: FilterTransactions::new(
&config.transactions,
&limit.transactions,
FilterTransactionsType::Transaction,
)?,
transactions_status: FilterTransactions::new(
&config.transactions_status,
&limit.transactions_status,
FilterTransactionsType::TransactionStatus,
)?,
entry: FilterEntry::new(&config.entry, &limit.entry)?,
blocks: FilterBlocks::new(&config.blocks, &limit.blocks)?,
blocks_meta: FilterBlocksMeta::new(&config.blocks_meta, &limit.blocks_meta)?,
@ -91,38 +101,42 @@ impl Filter {
}
pub fn get_filters<'a>(
&self,
&'a self,
message: &'a Message,
commitment: Option<CommitmentLevel>,
) -> Vec<(Vec<String>, MessageRef<'a>)> {
) -> Box<dyn Iterator<Item = (Vec<String>, MessageRef<'a>)> + Send + 'a> {
match message {
Message::Account(message) => self.accounts.get_filters(message),
Message::Slot(message) => self.slots.get_filters(message, commitment),
Message::Transaction(message) => self.transactions.get_filters(message),
Message::Transaction(message) => Box::new(
self.transactions
.get_filters(message)
.chain(self.transactions_status.get_filters(message)),
),
Message::Entry(message) => self.entry.get_filters(message),
Message::Block(message) => self.blocks.get_filters(message),
Message::BlockMeta(message) => self.blocks_meta.get_filters(message),
}
}
pub fn get_update(
&self,
message: &Message,
pub fn get_update<'a>(
&'a self,
message: &'a Message,
commitment: Option<CommitmentLevel>,
) -> Vec<SubscribeUpdate> {
self.get_filters(message, commitment)
.into_iter()
.filter_map(|(filters, message)| {
if filters.is_empty() {
None
} else {
Some(SubscribeUpdate {
filters,
update_oneof: Some(message.to_proto(&self.accounts_data_slice)),
})
}
})
.collect()
) -> Box<dyn Iterator<Item = SubscribeUpdate> + Send + 'a> {
Box::new(
self.get_filters(message, commitment)
.filter_map(|(filters, message)| {
if filters.is_empty() {
None
} else {
Some(SubscribeUpdate {
filters,
update_oneof: Some(message.to_proto(&self.accounts_data_slice)),
})
}
}),
)
}
pub fn get_pong_msg(&self) -> Option<SubscribeUpdate> {
@ -197,12 +211,18 @@ impl FilterAccounts {
Ok(required)
}
fn get_filters<'a>(&self, message: &'a MessageAccount) -> Vec<(Vec<String>, MessageRef<'a>)> {
fn get_filters<'a>(
&'a self,
message: &'a MessageAccount,
) -> Box<dyn Iterator<Item = (Vec<String>, MessageRef<'a>)> + Send + 'a> {
let mut filter = FilterAccountsMatch::new(self);
filter.match_account(&message.account.pubkey);
filter.match_owner(&message.account.owner);
filter.match_data(&message.account.data);
vec![(filter.get_filters(), MessageRef::Account(message))]
Box::new(std::iter::once((
filter.get_filters(),
MessageRef::Account(message),
)))
}
}
@ -391,11 +411,11 @@ impl FilterSlots {
}
fn get_filters<'a>(
&self,
&'a self,
message: &'a MessageSlot,
commitment: Option<CommitmentLevel>,
) -> Vec<(Vec<String>, MessageRef<'a>)> {
vec![(
) -> Box<dyn Iterator<Item = (Vec<String>, MessageRef<'a>)> + Send + 'a> {
Box::new(std::iter::once((
self.filters
.iter()
.filter_map(|(name, inner)| {
@ -407,10 +427,16 @@ impl FilterSlots {
})
.collect(),
MessageRef::Slot(message),
)]
)))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FilterTransactionsType {
Transaction,
TransactionStatus,
}
#[derive(Debug, Clone)]
pub struct FilterTransactionsInner {
vote: Option<bool>,
@ -421,8 +447,9 @@ pub struct FilterTransactionsInner {
account_required: Vec<Pubkey>,
}
#[derive(Debug, Default, Clone)]
#[derive(Debug, Clone)]
pub struct FilterTransactions {
filter_type: FilterTransactionsType,
filters: HashMap<String, FilterTransactionsInner>,
}
@ -430,10 +457,11 @@ impl FilterTransactions {
fn new(
configs: &HashMap<String, SubscribeRequestFilterTransactions>,
limit: &ConfigGrpcFiltersTransactions,
filter_type: FilterTransactionsType,
) -> anyhow::Result<Self> {
ConfigGrpcFilters::check_max(configs.len(), limit.max)?;
let mut this = Self::default();
let mut filters = HashMap::new();
for (name, filter) in configs {
ConfigGrpcFilters::check_any(
filter.vote.is_none()
@ -456,7 +484,7 @@ impl FilterTransactions {
limit.account_required_max,
)?;
this.filters.insert(
filters.insert(
name.clone(),
FilterTransactionsInner {
vote: filter.vote,
@ -485,13 +513,16 @@ impl FilterTransactions {
},
);
}
Ok(this)
Ok(Self {
filter_type,
filters,
})
}
pub fn get_filters<'a>(
&self,
&'a self,
message: &'a MessageTransaction,
) -> Vec<(Vec<String>, MessageRef<'a>)> {
) -> Box<dyn Iterator<Item = (Vec<String>, MessageRef<'a>)> + Send + 'a> {
let filters = self
.filters
.iter()
@ -565,7 +596,11 @@ impl FilterTransactions {
Some(name.clone())
})
.collect();
vec![(filters, MessageRef::Transaction(message))]
let message = match self.filter_type {
FilterTransactionsType::Transaction => MessageRef::Transaction(message),
FilterTransactionsType::TransactionStatus => MessageRef::TransactionStatus(message),
};
Box::new(std::iter::once((filters, message)))
}
}
@ -590,8 +625,14 @@ impl FilterEntry {
})
}
fn get_filters<'a>(&self, message: &'a MessageEntry) -> Vec<(Vec<String>, MessageRef<'a>)> {
vec![(self.filters.clone(), MessageRef::Entry(message))]
fn get_filters<'a>(
&'a self,
message: &'a MessageEntry,
) -> Box<dyn Iterator<Item = (Vec<String>, MessageRef<'a>)> + Send + 'a> {
Box::new(std::iter::once((
self.filters.clone(),
MessageRef::Entry(message),
)))
}
}
@ -654,68 +695,68 @@ impl FilterBlocks {
Ok(this)
}
fn get_filters<'a>(&self, message: &'a MessageBlock) -> Vec<(Vec<String>, MessageRef<'a>)> {
self.filters
.iter()
.map(|(filter, inner)| {
#[allow(clippy::unnecessary_filter_map)]
let transactions =
if matches!(inner.include_transactions, None | Some(true)) {
message
.transactions
.iter()
.filter_map(|tx| {
if !inner.account_include.is_empty()
&& tx.transaction.message().account_keys().iter().all(
|pubkey| {
inner.account_include.binary_search(pubkey).is_err()
},
)
{
return None;
}
fn get_filters<'a>(
&'a self,
message: &'a MessageBlock,
) -> Box<dyn Iterator<Item = (Vec<String>, MessageRef<'a>)> + Send + 'a> {
Box::new(self.filters.iter().map(move |(filter, inner)| {
#[allow(clippy::unnecessary_filter_map)]
let transactions = if matches!(inner.include_transactions, None | Some(true)) {
message
.transactions
.iter()
.filter_map(|tx| {
if !inner.account_include.is_empty()
&& tx
.transaction
.message()
.account_keys()
.iter()
.all(|pubkey| inner.account_include.binary_search(pubkey).is_err())
{
return None;
}
Some(tx)
})
.collect::<Vec<_>>()
} else {
vec![]
};
Some(tx)
})
.collect::<Vec<_>>()
} else {
vec![]
};
#[allow(clippy::unnecessary_filter_map)]
let accounts = if inner.include_accounts == Some(true) {
message
.accounts
.iter()
.filter_map(|account| {
if !inner.account_include.is_empty()
&& inner
.account_include
.binary_search(&account.pubkey)
.is_err()
{
return None;
}
#[allow(clippy::unnecessary_filter_map)]
let accounts = if inner.include_accounts == Some(true) {
message
.accounts
.iter()
.filter_map(|account| {
if !inner.account_include.is_empty()
&& inner
.account_include
.binary_search(&account.pubkey)
.is_err()
{
return None;
}
Some(account)
})
.collect::<Vec<_>>()
} else {
vec![]
};
Some(account)
})
.collect::<Vec<_>>()
} else {
vec![]
};
let entries = if inner.include_entries == Some(true) {
message.entries.iter().collect::<Vec<_>>()
} else {
vec![]
};
let entries = if inner.include_entries == Some(true) {
message.entries.iter().collect::<Vec<_>>()
} else {
vec![]
};
(
vec![filter.clone()],
MessageRef::Block((message, transactions, accounts, entries).into()),
)
})
.collect()
(
vec![filter.clone()],
MessageRef::Block((message, transactions, accounts, entries).into()),
)
}))
}
}
@ -740,8 +781,14 @@ impl FilterBlocksMeta {
})
}
fn get_filters<'a>(&self, message: &'a MessageBlockMeta) -> Vec<(Vec<String>, MessageRef<'a>)> {
vec![(self.filters.clone(), MessageRef::BlockMeta(message))]
fn get_filters<'a>(
&'a self,
message: &'a MessageBlockMeta,
) -> Box<dyn Iterator<Item = (Vec<String>, MessageRef<'a>)> + Send + 'a> {
Box::new(std::iter::once((
self.filters.clone(),
MessageRef::BlockMeta(message),
)))
}
}
@ -788,7 +835,7 @@ mod tests {
crate::{
config::ConfigGrpcFilters,
filters::Filter,
grpc::{Message, MessageTransaction, MessageTransactionInfo},
grpc::{Message, MessageRef, MessageTransaction, MessageTransactionInfo},
},
solana_sdk::{
hash::Hash,
@ -855,6 +902,7 @@ mod tests {
accounts: HashMap::new(),
slots: HashMap::new(),
transactions: HashMap::new(),
transactions_status: HashMap::new(),
blocks: HashMap::new(),
blocks_meta: HashMap::new(),
entry: HashMap::new(),
@ -884,6 +932,7 @@ mod tests {
accounts,
slots: HashMap::new(),
transactions: HashMap::new(),
transactions_status: HashMap::new(),
blocks: HashMap::new(),
blocks_meta: HashMap::new(),
entry: HashMap::new(),
@ -918,6 +967,7 @@ mod tests {
accounts: HashMap::new(),
slots: HashMap::new(),
transactions,
transactions_status: HashMap::new(),
blocks: HashMap::new(),
blocks_meta: HashMap::new(),
entry: HashMap::new(),
@ -951,6 +1001,7 @@ mod tests {
accounts: HashMap::new(),
slots: HashMap::new(),
transactions,
transactions_status: HashMap::new(),
blocks: HashMap::new(),
blocks_meta: HashMap::new(),
entry: HashMap::new(),
@ -990,6 +1041,7 @@ mod tests {
accounts: HashMap::new(),
slots: HashMap::new(),
transactions,
transactions_status: HashMap::new(),
blocks: HashMap::new(),
blocks_meta: HashMap::new(),
entry: HashMap::new(),
@ -1003,9 +1055,12 @@ mod tests {
let message_transaction =
create_message_transaction(&keypair_b, vec![account_key_b, account_key_a]);
let message = Message::Transaction(message_transaction);
for (filters, _message) in filter.get_filters(&message, None) {
assert!(!filters.is_empty());
}
let updates = filter.get_filters(&message, None).collect::<Vec<_>>();
assert_eq!(updates.len(), 2);
assert_eq!(updates[0].0, vec!["serum"]);
assert!(matches!(updates[0].1, MessageRef::Transaction(_)));
assert_eq!(updates[1].0, Vec::<String>::new());
assert!(matches!(updates[1].1, MessageRef::TransactionStatus(_)));
}
#[test]
@ -1033,6 +1088,7 @@ mod tests {
accounts: HashMap::new(),
slots: HashMap::new(),
transactions,
transactions_status: HashMap::new(),
blocks: HashMap::new(),
blocks_meta: HashMap::new(),
entry: HashMap::new(),
@ -1046,9 +1102,12 @@ mod tests {
let message_transaction =
create_message_transaction(&keypair_b, vec![account_key_b, account_key_a]);
let message = Message::Transaction(message_transaction);
for (filters, _message) in filter.get_filters(&message, None) {
assert!(!filters.is_empty());
}
let updates = filter.get_filters(&message, None).collect::<Vec<_>>();
assert_eq!(updates.len(), 2);
assert_eq!(updates[0].0, vec!["serum"]);
assert!(matches!(updates[0].1, MessageRef::Transaction(_)));
assert_eq!(updates[1].0, Vec::<String>::new());
assert!(matches!(updates[1].1, MessageRef::TransactionStatus(_)));
}
#[test]
@ -1076,6 +1135,7 @@ mod tests {
accounts: HashMap::new(),
slots: HashMap::new(),
transactions,
transactions_status: HashMap::new(),
blocks: HashMap::new(),
blocks_meta: HashMap::new(),
entry: HashMap::new(),
@ -1125,6 +1185,7 @@ mod tests {
accounts: HashMap::new(),
slots: HashMap::new(),
transactions,
transactions_status: HashMap::new(),
blocks: HashMap::new(),
blocks_meta: HashMap::new(),
entry: HashMap::new(),
@ -1140,9 +1201,12 @@ mod tests {
vec![account_key_x, account_key_y, account_key_z],
);
let message = Message::Transaction(message_transaction);
for (filters, _message) in filter.get_filters(&message, None) {
assert!(!filters.is_empty());
}
let updates = filter.get_filters(&message, None).collect::<Vec<_>>();
assert_eq!(updates.len(), 2);
assert_eq!(updates[0].0, vec!["serum"]);
assert!(matches!(updates[0].1, MessageRef::Transaction(_)));
assert_eq!(updates[1].0, Vec::<String>::new());
assert!(matches!(updates[1].1, MessageRef::TransactionStatus(_)));
}
#[test]
@ -1176,6 +1240,7 @@ mod tests {
accounts: HashMap::new(),
slots: HashMap::new(),
transactions,
transactions_status: HashMap::new(),
blocks: HashMap::new(),
blocks_meta: HashMap::new(),
entry: HashMap::new(),

18
yellowstone-grpc-geyser/src/grpc.rs
View File

@ -51,6 +51,7 @@ use {
SubscribeUpdateAccount, SubscribeUpdateAccountInfo, SubscribeUpdateBlock,
SubscribeUpdateBlockMeta, SubscribeUpdateEntry, SubscribeUpdatePing,
SubscribeUpdateSlot, SubscribeUpdateTransaction, SubscribeUpdateTransactionInfo,
SubscribeUpdateTransactionStatus, TransactionError as SubscribeUpdateTransactionError,
},
},
};
@ -393,6 +394,7 @@ pub enum MessageRef<'a> {
Slot(&'a MessageSlot),
Account(&'a MessageAccount),
Transaction(&'a MessageTransaction),
TransactionStatus(&'a MessageTransaction),
Entry(&'a MessageEntry),
Block(MessageBlockRef<'a>),
BlockMeta(&'a MessageBlockMeta),
@ -415,6 +417,21 @@ impl<'a> MessageRef<'a> {
transaction: Some(message.transaction.to_proto()),
slot: message.slot,
}),
Self::TransactionStatus(message) => {
UpdateOneof::TransactionStatus(SubscribeUpdateTransactionStatus {
slot: message.slot,
signature: message.transaction.signature.as_ref().into(),
is_vote: message.transaction.is_vote,
index: message.transaction.index as u64,
err: match &message.transaction.meta.status {
Ok(()) => None,
Err(err) => Some(SubscribeUpdateTransactionError {
err: bincode::serialize(&err)
.expect("transaction error to serialize to bytes"),
}),
},
})
}
Self::Entry(message) => UpdateOneof::Entry(message.to_proto()),
Self::Block(message) => UpdateOneof::Block(SubscribeUpdateBlock {
slot: message.slot,
@ -1204,6 +1221,7 @@ impl Geyser for GrpcService {
accounts: HashMap::new(),
slots: HashMap::new(),
transactions: HashMap::new(),
transactions_status: HashMap::new(),
blocks: HashMap::new(),
blocks_meta: HashMap::new(),
entry: HashMap::new(),

10
yellowstone-grpc-proto/proto/geyser.proto
View File

@ -26,6 +26,7 @@ message SubscribeRequest {
map<string, SubscribeRequestFilterAccounts> accounts = 1;
map<string, SubscribeRequestFilterSlots> slots = 2;
map<string, SubscribeRequestFilterTransactions> transactions = 3;
map<string, SubscribeRequestFilterTransactions> transactions_status = 10;
map<string, SubscribeRequestFilterBlocks> blocks = 4;
map<string, SubscribeRequestFilterBlocksMeta> blocks_meta = 5;
map<string, SubscribeRequestFilterEntry> entry = 8;
@ -96,6 +97,7 @@ message SubscribeUpdate {
SubscribeUpdateAccount account = 2;
SubscribeUpdateSlot slot = 3;
SubscribeUpdateTransaction transaction = 4;
SubscribeUpdateTransactionStatus transaction_status = 10;
SubscribeUpdateBlock block = 5;
SubscribeUpdatePing ping = 6;
SubscribeUpdatePong pong = 9;
@ -140,6 +142,14 @@ message SubscribeUpdateTransactionInfo {
uint64 index = 5;
}
message SubscribeUpdateTransactionStatus {
uint64 slot = 1;
bytes signature = 2;
bool is_vote = 3;
uint64 index = 4;
solana.storage.ConfirmedBlock.TransactionError err = 5;
}
message SubscribeUpdateBlock {
uint64 slot = 1;
string blockhash = 2;

3
yellowstone-grpc-tools/src/bin/grpc-google-pubsub.rs
View File

@ -282,6 +282,9 @@ impl ArgsAction {
Some(UpdateOneof::Transaction(msg)) => {
info!("#{}, transaction", msg.slot)
}
Some(UpdateOneof::TransactionStatus(msg)) => {
info!("#{}, transaction status", msg.slot)
}
Some(UpdateOneof::Block(msg)) => info!("#{}, block", msg.slot),
Some(UpdateOneof::Ping(_)) => {}
Some(UpdateOneof::Pong(_)) => {}

1
yellowstone-grpc-tools/src/bin/grpc-kafka.rs
View File

@ -275,6 +275,7 @@ impl ArgsAction {
UpdateOneof::Account(msg) => msg.slot,
UpdateOneof::Slot(msg) => msg.slot,
UpdateOneof::Transaction(msg) => msg.slot,
UpdateOneof::TransactionStatus(msg) => msg.slot,
UpdateOneof::Block(msg) => msg.slot,
UpdateOneof::Ping(_) => continue,
UpdateOneof::Pong(_) => continue,

2
yellowstone-grpc-tools/src/config.rs
View File

@ -44,6 +44,7 @@ pub struct ConfigGrpcRequest {
pub slots: HashMap<String, ConfigGrpcRequestSlots>,
pub accounts: HashMap<String, ConfigGrpcRequestAccounts>,
pub transactions: HashMap<String, ConfigGrpcRequestTransactions>,
pub transactions_status: HashMap<String, ConfigGrpcRequestTransactions>,
pub entries: HashSet<String>,
pub blocks: HashMap<String, ConfigGrpcRequestBlocks>,
pub blocks_meta: HashSet<String>,
@ -71,6 +72,7 @@ impl GrpcRequestToProto<SubscribeRequest> for ConfigGrpcRequest {
slots: ConfigGrpcRequest::map_to_proto(self.slots),
accounts: ConfigGrpcRequest::map_to_proto(self.accounts),
transactions: ConfigGrpcRequest::map_to_proto(self.transactions),
transactions_status: ConfigGrpcRequest::map_to_proto(self.transactions_status),
entry: ConfigGrpcRequest::set_to_proto(self.entries),
blocks: ConfigGrpcRequest::map_to_proto(self.blocks),
blocks_meta: ConfigGrpcRequest::set_to_proto(self.blocks_meta),

3
yellowstone-grpc-tools/src/prom.rs
View File

@ -111,6 +111,7 @@ pub enum GprcMessageKind {
Account,
Slot,
Transaction,
TransactionStatus,
Block,
Ping,
Pong,
@ -125,6 +126,7 @@ impl From<&UpdateOneof> for GprcMessageKind {
UpdateOneof::Account(_) => Self::Account,
UpdateOneof::Slot(_) => Self::Slot,
UpdateOneof::Transaction(_) => Self::Transaction,
UpdateOneof::TransactionStatus(_) => Self::TransactionStatus,
UpdateOneof::Block(_) => Self::Block,
UpdateOneof::Ping(_) => Self::Ping,
UpdateOneof::Pong(_) => Self::Pong,
@ -140,6 +142,7 @@ impl GprcMessageKind {
GprcMessageKind::Account => "account",
GprcMessageKind::Slot => "slot",
GprcMessageKind::Transaction => "transaction",
GprcMessageKind::TransactionStatus => "transactionstatus",
GprcMessageKind::Block => "block",
GprcMessageKind::Ping => "ping",
GprcMessageKind::Pong => "pong",