Adding custom tpu send transaction example #380
Conversation
| @@ -122,6 +122,15 @@ pub struct TransactionService { | |||
|
|
|||
| impl TransactionService { | |||
| pub async fn send_transaction( | |||
| &self, | |||
| tx: Transaction, | |||
| max_retries: Option<u16>, | |||
There was a problem hiding this comment.
nit: we should stick to one data type for this (i.e. usize)
godmodegalactus
changed the title
| @@ -127,7 +128,7 @@ impl TpuService { | |||
| .get_slot_leaders(current_slot, last_slot) | |||
| .await?; | |||
| // get next leader with its tpu port | |||
| let connections_to_keep = next_leaders | |||
| let connections_to_keep: HashMap<_, _> = next_leaders | |||
There was a problem hiding this comment.
why do you need that?
maybe you could specify the full type in that case
There was a problem hiding this comment.
There is no method to collect as Vec. Specifying HashMap should be fine here.
| @@ -11,10 +11,10 @@ pub fn poll_cluster_info( | |||
| ) -> AnyhowJoinHandle { | |||
| // task MUST not terminate but might be aborted from outside | |||
| tokio::spawn(async move { | |||
| tokio::time::sleep(Duration::from_secs(1)).await; | |||
There was a problem hiding this comment.
pls add a comment why you need that
| @@ -23,7 +23,7 @@ pub fn poll_cluster_info( | |||
| Err(error) => { | |||
| warn!("rpc_cluster_info failed <{:?}> - retrying", error); | |||
| // throttle | |||
| tokio::time::sleep(Duration::from_secs(2500)).await; | |||
| tokio::time::sleep(Duration::from_secs(10)).await; | |||
There was a problem hiding this comment.
would put that on a bugfix branch
core/src/utils.rs
Outdated
| started.elapsed().as_secs_f32() * 1000.0 | ||
| ); | ||
| } | ||
| Ok(Err(_error)) => { |
There was a problem hiding this comment.
!! you should handle Lagged and Closed are differently:
- lagged: continue
- Closed: sleep some 10 second, then panic (create_grpc_multiplex_blocks_subscription reconnect_loop never terminates unless cluster_endpoint_tasks get aborted which happens on shutdown; this method could gracefully wait if it detects shutdown)
There was a problem hiding this comment.
The aim of this method is to wait till we get block of required commitment, this channel should never lag. If it lags then the logic is wrong.
| } | ||
| Err(_elapsed) => { | ||
| debug!( | ||
| "waiting for latest block info ({}) ... {:.02}ms", |
There was a problem hiding this comment.
pls add to log line what it's doing -> i.e. "timed out waiting for latest block info (...) - continue waiting"
There was a problem hiding this comment.
This is just for debugging we do not need any additional log lines.
| loop { | ||
| match timeout(Duration::from_millis(500), blockinfo_stream.recv()).await { | ||
| Ok(Ok(block_info)) => { | ||
| if block_info.commitment_config == commitment_config { |
There was a problem hiding this comment.
would you accept block_info(finalized) if you request confirmed or not?
| @@ -7,7 +9,7 @@ pub type WireTransaction = Vec<u8>; | |||
| pub struct SentTransactionInfo { | |||
There was a problem hiding this comment.
why is it called Sent? this reads to me like it happened already but the actually that struct contains the transaction to be sent
There was a problem hiding this comment.
Yes it is structure that trackts the transaction that is sent.
| @@ -7,7 +9,7 @@ pub type WireTransaction = Vec<u8>; | |||
| pub struct SentTransactionInfo { | |||
| pub signature: Signature, | |||
| pub slot: Slot, | |||
| pub transaction: WireTransaction, | |||
| pub transaction: Arc<WireTransaction>, | |||
There was a problem hiding this comment.
could that Arc be avoided somehow? let me try for a minute
| nb_signers, signer_balance | ||
| ); | ||
|
|
||
| let validator_identity = Arc::new( |
There was a problem hiding this comment.
you could move the ValidatorIdentity out of quic-proxy which wraps that case explicitly
| endpoints.vote_account_notifier, | ||
| ); | ||
|
|
||
| let count = args.transaction_count.unwrap_or(10); |
There was a problem hiding this comment.
nit: should put that in Args struct
| let count = args.transaction_count.unwrap_or(10); | ||
| let prioritization_heap_size = Some(count * args.number_of_seconds); | ||
|
|
||
| let tpu_config = TpuServiceConfig { |
There was a problem hiding this comment.
suggest to print TpuServiceConfig values on console to help the user to inspect the parameters used
| let tx = tx.clone(); | ||
| let signer = *signer; | ||
| tokio::spawn( | ||
| async move { (signer, rpc_client.send_and_confirm_transaction(&tx).await) }, |
There was a problem hiding this comment.
pls add .context
| ) | ||
| .await?; | ||
| let transaction_service_builder = TransactionServiceBuilder::new( | ||
| TxSender::new(data_cache.clone(), tpu_service.clone()), |
There was a problem hiding this comment.
the services are quite tangled. shouldn't it be easy to just wire up the TxSender? it requires recent blockhash and leader schedule
leader schedule could be:
- full schedule from RPC
- schedule for current epoch as text-file with each lead qualified by slots
- hardcoded leaders (if the setup is just 1 test-validator or a small number of test nodes)
No description provided.