Skip to content

Commit 2857ab8

Browse files
leafaarleafaar
committed
feat: added logs for debug and improved gentx
1 parent 372e17b commit 2857ab8

8 files changed

+143
-61
lines changed

‎Cargo.lock

-5
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎Cargo.toml

+3-3
Original file line numberDiff line numberDiff line change
@@ -99,6 +99,6 @@ solana-derivation-path = { git = "https://github.com/rpcpool/solana-public.git",
9999
solana-feature-set = { git = "https://github.com/rpcpool/solana-public.git", tag = "v2.1.11-triton-public" }
100100

101101
# https://github.com/anza-xyz/agave/blob/v2.0.4/Cargo.toml#L502-L533
102-
[patch.crates-io.curve25519-dalek]
103-
git = "https://github.com/anza-xyz/curve25519-dalek.git"
104-
rev = "b500cdc2a920cd5bff9e2dd974d7b97349d61464"
102+
# [patch.crates-io.curve25519-dalek]
103+
# git = "https://github.com/anza-xyz/curve25519-dalek.git"
104+
# rev = "b500cdc2a920cd5bff9e2dd974d7b97349d61464"

‎src/bin/gentx.rs

+14-3
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,10 @@ struct Args {
6161
/// Check the balance of the wallet and airdrop Solana (devnet and testnet only)
6262
#[clap(long)]
6363
pub airdrop: bool, // default is false
64+
65+
/// Use legacy payload format
66+
#[clap(long)]
67+
pub legacy: bool, // default is the new tx payload
6468
}
6569

6670
#[derive(Debug, Clone, Deserialize)]
@@ -101,9 +105,10 @@ struct Config {
101105

102106
impl Config {
103107
pub async fn load(path: impl AsRef<Path>) -> anyhow::Result<Self> {
108+
let path = path.as_ref();
104109
let contents = fs::read(path)
105110
.await
106-
.with_context(|| "failed to read config")?;
111+
.with_context(|| format!("failed to read config from {:?}", path))?;
107112
Ok(serde_yaml::from_slice(&contents)?)
108113
}
109114

@@ -183,6 +188,7 @@ impl TransactionSender {
183188
&self,
184189
transaction: VersionedTransaction,
185190
config: RpcSendTransactionConfigWithBlockList,
191+
should_use_legacy_txn: bool,
186192
) -> anyhow::Result<Signature> {
187193
match self {
188194
Self::Jet { rpc } => rpc
@@ -191,7 +197,11 @@ impl TransactionSender {
191197
.map_err(Into::into),
192198
Self::JetGateway { tx } => {
193199
let signature = transaction.signatures[0];
194-
let payload = TransactionPayload::try_from((&transaction, config))?;
200+
let payload = if should_use_legacy_txn {
201+
TransactionPayload::to_legacy(&transaction, &config)?
202+
} else {
203+
TransactionPayload::try_from((&transaction, config))?
204+
};
195205
let proto_tx = payload.to_proto::<PublishTransaction>();
196206
tx.lock()
197207
.await
@@ -272,6 +282,7 @@ async fn main() -> anyhow::Result<()> {
272282
latest_slot,
273283
latest_block.block_height.unwrap_or(0)
274284
);
285+
let should_use_legacy_txn = args.legacy;
275286

276287
let landed = Arc::new(AtomicUsize::new(0));
277288
let sender = Arc::new(TransactionSender::from_config(config.output.clone()).await?);
@@ -322,7 +333,7 @@ async fn main() -> anyhow::Result<()> {
322333
.filter_map(|addr| Pubkey::from_str(addr).ok())
323334
.collect(),
324335
};
325-
match sender.send(transaction, config).await {
336+
match sender.send(transaction, config, should_use_legacy_txn).await {
326337
Ok(send_signature) => {
327338
anyhow::ensure!(signature == send_signature, "received invalid signature from sender");
328339
info!("successfully send transaction {signature}");

‎src/cluster_tpu_info.rs

+65-49
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ use {
2626
sync::{broadcast, RwLock},
2727
time::{sleep, Duration, Instant},
2828
},
29-
tracing::{info, warn},
29+
tracing::{debug, info, warn},
3030
};
3131

3232
#[derive(Debug, Clone, Copy)]
@@ -484,50 +484,52 @@ impl BlocklistUpdater for LeadersSelector {
484484

485485
#[async_trait::async_trait]
486486
pub trait BlockLeaders {
487-
/// Takes a mutable reference of a vector containing TpuInfo and retain only those allowed by the
488-
/// lists indicated by blocklist_keys
489-
/// # Example
490-
/// ```
491-
///
492-
/// struct BlocklistYellowstone {
493-
/// blocklist: HashMap<Pubkey, HashSet<Pubkey>>
494-
/// }
495-
///
496-
///
497-
/// #[async_trait::async_trait]
498-
/// impl BlockLeaders for Blocklists {
499-
/// async fn block_leaders(&self, tpus: &mut Vec<TpuInfo>, blocklist_keys: &[Pubkey]) {
500-
/// let blocklist = self.blocklist;
501-
///
502-
/// tpus.retain(|info| {
503-
/// for key_list in blocklist_keys.iter() {
504-
/// if let Some(blocklist_hash) = blocklist.get(key_list) {
505-
/// if blocklist_hash.contains(&info.leader) {
506-
/// return false;
507-
/// }
508-
/// }
509-
/// true
510-
/// }
511-
/// }
512-
/// )
513-
/// }
514-
/// }
515-
///
516-
///
517-
/// async fn main() {
518-
/// let key1 = Pubkey::new_unique();
519-
/// let mut tpus = vec![TpuInfo {
520-
/// leader: key1,
521-
/// ...
522-
/// }]
523-
/// let blocklist_yellowstone = {
524-
/// blocklist: hashmap!{key1 => hashset!{key1}}
525-
/// }
526-
/// blocklist_yellowstone.block_leaders(&mut tpus, &[key1]).await;
527-
/// // Now tpus should have a length of zero
528-
/// assert!(tpus.is_empty());
529-
/// }
530-
/// ```
487+
/*
488+
Takes a mutable reference of a vector containing TpuInfo and retain only those allowed by the
489+
lists indicated by blocklist_keys
490+
# Example
491+
```
492+
493+
struct BlocklistYellowstone {
494+
blocklist: HashMap<Pubkey, HashSet<Pubkey>>
495+
}
496+
497+
498+
#[async_trait::async_trait]
499+
impl BlockLeaders for Blocklists {
500+
async fn block_leaders(&self, tpus: &mut Vec<TpuInfo>, blocklist_keys: &[Pubkey]) {
501+
let blocklist = self.blocklist;
502+
503+
tpus.retain(|info| {
504+
for key_list in blocklist_keys.iter() {
505+
if let Some(blocklist_hash) = blocklist.get(key_list) {
506+
if blocklist_hash.contains(&info.leader) {
507+
return false;
508+
}
509+
}
510+
true
511+
}
512+
}
513+
)
514+
}
515+
}
516+
517+
518+
async fn main() {
519+
let key1 = Pubkey::new_unique();
520+
let mut tpus = vec![TpuInfo {
521+
leader: key1,
522+
...
523+
}]
524+
let blocklist_yellowstone = {
525+
blocklist: hashmap!{key1 => hashset!{key1}}
526+
}
527+
blocklist_yellowstone.block_leaders(&mut tpus, &[key1]).await;
528+
// Now tpus should have a length of zero
529+
assert!(tpus.is_empty());
530+
}
531+
```
532+
*/
531533
async fn block_leaders(&self, tpus: &mut Vec<TpuInfo>, blocklist_keys: &[Pubkey]);
532534
}
533535

@@ -541,27 +543,41 @@ impl BlockLeaders for LeadersSelector {
541543

542544
tpus.retain(|info| {
543545
let mut is_allow = false;
546+
let leader = info.leader;
544547

545-
for key_list in blocklist_keys.iter() {
548+
// Check deny lists first
549+
for key_list in blocklist_keys {
546550
if let Some(deny_hash) = deny_lists.get(key_list) {
547-
if deny_hash.contains(&info.leader) {
551+
if deny_hash.contains(&leader) {
548552
blocklisted += 1;
553+
debug!(
554+
"Leader {} BLOCKED - found in deny list {}",
555+
leader, key_list
556+
);
549557
return false;
550558
}
551559
}
560+
}
552561

562+
// Then check allow lists
563+
for key_list in blocklist_keys {
553564
if let Some(allow_hash) = allow_lists.get(key_list) {
554-
if !allow_hash.contains(&info.leader) {
565+
if !allow_hash.contains(&leader) {
555566
blocklisted += 1;
567+
debug!(
568+
"Leader {} BLOCKED - not found in allow list {}",
569+
leader, key_list
570+
);
556571
return false;
557572
} else {
558573
is_allow = true;
559574
}
560575
}
561576
}
562577

563-
if !is_allow && self.blocklist.contains(&info.leader) {
578+
if !is_allow && self.blocklist.contains(&leader) {
564579
blocklisted += 1;
580+
debug!("Leader {} BLOCKED - found in global blocklist", leader);
565581
return false;
566582
}
567583
true

‎src/grpc_geyser.rs

+3
Original file line numberDiff line numberDiff line change
@@ -202,6 +202,7 @@ impl GeyserSubscriber {
202202
Some(Ok(msg)) => match msg.update_oneof {
203203
Some(UpdateOneof::Account(SubscribeUpdateAccount{account, ..})) => {
204204
if let Some(acc) = account {
205+
// todo: add a verification to ensure the owner is the blocklist program
205206
let pubkey_bytes = acc.pubkey;
206207

207208
if pubkey_bytes.len() == 32 {
@@ -395,6 +396,8 @@ impl GeyserSubscriber {
395396
}
396397
.and_then(|builder| async {
397398
builder
399+
// todo: we might want to consider reducing this since we are not using this much
400+
// for transactions/slots
398401
.max_decoding_message_size(128 * 1024 * 1024) // 128MiB, BlockMeta with rewards can be bigger than 60MiB
399402
.connect_timeout(Duration::from_secs(3))
400403
.timeout(Duration::from_secs(3))

‎src/payload.rs

+35-1
Original file line numberDiff line numberDiff line change
@@ -127,6 +127,41 @@ impl TryFrom<(&VersionedTransaction, RpcSendTransactionConfigWithBlockList)>
127127
RpcSendTransactionConfigWithBlockList,
128128
),
129129
) -> Result<Self, Self::Error> {
130+
let tx_bytes = bincode::serialize(transaction)?;
131+
132+
Ok(Self::New(TransactionWrapper {
133+
transaction: tx_bytes,
134+
config: Some(TransactionConfig {
135+
max_retries: config_with_blocklist
136+
.config
137+
.as_ref()
138+
.and_then(|c| c.max_retries.map(|r| r as u32)),
139+
blocklist_pdas: config_with_blocklist
140+
.blocklist_pdas
141+
.iter()
142+
.map(|p| p.to_string())
143+
.collect(),
144+
skip_preflight: config_with_blocklist
145+
.config
146+
.as_ref()
147+
.map(|c| c.skip_preflight)
148+
.unwrap_or_default(),
149+
skip_sanitize: config_with_blocklist
150+
.config
151+
.as_ref()
152+
.map(|c| c.skip_sanitize)
153+
.unwrap_or_default(),
154+
}),
155+
timestamp: Some(ms_since_epoch()),
156+
}))
157+
}
158+
}
159+
160+
impl TransactionPayload {
161+
pub fn to_legacy(
162+
transaction: &VersionedTransaction,
163+
config_with_blocklist: &RpcSendTransactionConfigWithBlockList,
164+
) -> Result<Self, PayloadError> {
130165
let encoding = config_with_blocklist
131166
.config
132167
.as_ref()
@@ -144,7 +179,6 @@ impl TryFrom<(&VersionedTransaction, RpcSendTransactionConfigWithBlockList)>
144179
_ => BASE64_STANDARD.encode(tx_bytes),
145180
};
146181

147-
// Create a new legacy payload with the config preserved
148182
Ok(Self::Legacy(LegacyPayload {
149183
transaction: tx_str,
150184
config: config_with_blocklist.config.unwrap_or_default(),

‎src/quic.rs

+7
Original file line numberDiff line numberDiff line change
@@ -221,12 +221,19 @@ impl QuicClient {
221221
.upcoming_leader_schedule
222222
.get_leader_tpus(leader_forward_count)
223223
.await;
224+
debug!(
225+
"Attempting to send to {} leaders before filtering",
226+
tpus_info.len()
227+
);
228+
224229
tpus_info.extend(self.extra_tpu_forward.iter().cloned());
225230

226231
self.leaders_selector
227232
.block_leaders(&mut tpus_info, &blocklist_keys)
228233
.await;
229234

235+
debug!("After filtering, sending to {} leaders", tpus_info.len());
236+
230237
let futs = tpus_info
231238
.into_iter()
232239
.filter_map(|tpu_info| {

‎src/transactions.rs

+16
Original file line numberDiff line numberDiff line change
@@ -276,6 +276,10 @@ impl SendTransactionsPool {
276276
}
277277

278278
pub fn send_transaction(&self, request: SendTransactionRequest) -> anyhow::Result<()> {
279+
debug!(
280+
"Sending transaction with signature {} and blocklist: {:?}",
281+
request.signature, request.blocklist_pdas
282+
);
279283
anyhow::ensure!(
280284
self.new_transactions_tx.send(request).is_ok(),
281285
"send service task finished"
@@ -415,6 +419,10 @@ impl TxChannel for QuicClient {
415419
leader_foward_count: usize,
416420
blocklist_keys: Vec<Pubkey>,
417421
) -> Option<BoxedTxChannelPermit> {
422+
debug!(
423+
"QuicClient::reserve called with blocklist: {:?}",
424+
blocklist_keys
425+
);
418426
QuicClient::reserve_send_permit(self, leader_foward_count, blocklist_keys)
419427
.await
420428
.map(BoxedTxChannelPermit::new)
@@ -756,10 +764,18 @@ impl SendTransactionsPoolTask {
756764
wire_transaction: Arc<Vec<u8>>,
757765
blocklist_keys: Vec<Pubkey>,
758766
) {
767+
debug!(
768+
"Spawning connect for tx {}, id {}, with blocklist: {:?}",
769+
signature, id, blocklist_keys
770+
);
759771
let leader_forward_count = self.config.leader_forward_count;
760772
let tx_channel = Arc::clone(&self.tx_channel);
761773
let blocklist_keys_clone = blocklist_keys.clone();
762774
let abort_handle = self.connecting_tasks.spawn(async move {
775+
debug!(
776+
"Reserving tx {} with blocklist: {:?}",
777+
signature, blocklist_keys_clone
778+
);
763779
tx_channel
764780
.reserve(leader_forward_count, blocklist_keys_clone)
765781
.await

0 commit comments

Comments
 (0)