feat: send transactions directly to leader TPU#1419
Conversation
📝 WalkthroughWalkthrough
Suggested reviewers: ✨ Finishing Touches🧪 Generate unit tests (beta)
⚔️ Resolve merge conflicts
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@magicblock-rpc-client/src/lib.rs`:
- Around line 344-373: Replace the single-shot OnceCell caching in tpu_sender
with a synchronized retry strategy that does not permanently retain None after
TpuClient::new fails. Track the cached result and retry eligibility using a
cooldown/backoff or TTL, while reusing a successfully initialized TpuSendFn
until refresh is needed; update the associated documentation to describe the
chosen retry behavior and failure semantics.
- Around line 696-727: Bound the TPU delivery future in send_transaction with a
tokio::time::timeout using the appropriate existing timeout duration or a
clearly defined bounded value. Handle timeout expiration as an unsuccessful TPU
send while preserving the current serialization, sender lookup, RPC error
handling, and signature fallback behavior; ensure a stalled
tpu_sender(wire_transaction) cannot delay completion after RPC succeeds.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: dbd529c4-ffe3-43d6-9668-cbb0ada2e1c3
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (4)
.agents/context/crates/magicblock-rpc-client.mdCargo.tomlmagicblock-rpc-client/Cargo.tomlmagicblock-rpc-client/src/lib.rs
| async fn tpu_sender(&self) -> Option<&TpuSendFn> { | ||
| self.tpu_sender | ||
| .get_or_init(|| async { | ||
| let websocket_url = self.websocket_url.as_deref()?; | ||
| match TpuClient::new( | ||
| "magicblock-rpc-client", | ||
| self.client.clone(), | ||
| websocket_url, | ||
| TpuClientConfig::default(), | ||
| ) | ||
| .await | ||
| { | ||
| Ok(client) => { | ||
| let client = Arc::new(client); | ||
| Some(Arc::new(move |wire_transaction| { | ||
| let client = client.clone(); | ||
| Box::pin(async move { | ||
| client.send_wire_transaction(wire_transaction).await | ||
| }) as TpuSendFuture<'static> | ||
| }) as TpuSendFn) | ||
| } | ||
| Err(err) => { | ||
| warn!(?err, "Failed to initialize leader TPU client; continuing with RPC delivery"); | ||
| None | ||
| } | ||
| } | ||
| }) | ||
| .await | ||
| .as_ref() | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
TPU sender never retries after the first initialization failure.
tpu_sender() caches the result of TpuClient::new(...) in an Arc<OnceCell<Option<TpuSendFn>>>. If the very first call fails (e.g. a transient websocket blip while MagicblockRpcClient is constructed at startup), the cell permanently stores None, and every subsequent send_transaction call silently skips TPU delivery for the entire lifetime of this client instance — with no re-attempt, backoff, or periodic refresh. Given this client is designed to be cheaply cloned and long-lived, a single transient failure disables the whole feature for the process's lifetime. The doc (.agents/context/crates/magicblock-rpc-client.md lines 158-162) says failure is "non-fatal" but doesn't mention this permanence, which could mislead operators into expecting recovery once the websocket comes back.
Consider a retry-with-backoff/TTL re-attempt strategy (e.g., store (Instant, Option<TpuSendFn>) behind a mutex and re-run the initializer after a cooldown) instead of a single-shot OnceCell, and update the doc to reflect whichever behavior is intended.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@magicblock-rpc-client/src/lib.rs` around lines 344 - 373, Replace the
single-shot OnceCell caching in tpu_sender with a synchronized retry strategy
that does not permanently retain None after TpuClient::new fails. Track the
cached result and retry eligibility using a cooldown/backoff or TTL, while
reusing a successfully initialized TpuSendFn until refresh is needed; update the
associated documentation to describe the chosen retry behavior and failure
semantics.
| let wire_transaction = bincode::serialize(tx).ok(); | ||
| let rpc_send = self | ||
| .client | ||
| .send_transaction_with_config(tx, SEND_TRANSACTION_CONFIG) | ||
| .await | ||
| .map_err(|e| { | ||
| MagicBlockRpcClientError::SendTransaction(Box::new(e)) | ||
| })?; | ||
| .send_transaction_with_config(tx, SEND_TRANSACTION_CONFIG); | ||
| // Type-erase this future so every generic transaction caller does not | ||
| // inherit the deeply nested TPU client future in its async state machine. | ||
| let tpu_send: TpuSendFuture<'_> = Box::pin(async { | ||
| let Some(wire_transaction) = wire_transaction else { | ||
| warn!( | ||
| "Failed to serialize transaction for leader TPU delivery" | ||
| ); | ||
| return false; | ||
| }; | ||
| let Some(tpu_sender) = self.tpu_sender().await else { | ||
| return false; | ||
| }; | ||
| tpu_sender(wire_transaction).await | ||
| }); | ||
|
|
||
| let (rpc_result, tpu_sent) = tokio::join!(rpc_send, tpu_send); | ||
| let sig = match rpc_result { | ||
| Ok(signature) => signature, | ||
| Err(err) if tpu_sent => { | ||
| warn!(?err, "RPC transaction delivery failed after leader TPU delivery succeeded"); | ||
| *tx.get_signature() | ||
| } | ||
| Err(err) => { | ||
| return Err(MagicBlockRpcClientError::SendTransaction( | ||
| Box::new(err), | ||
| )); | ||
| } | ||
| }; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win
tokio::join! can block send_transaction on a hung/slow TPU delivery even when RPC already succeeded.
tokio::join!(rpc_send, tpu_send) only completes once both futures resolve. If tpu_sender(wire_transaction).await (i.e. TpuClient::send_wire_transaction) hangs or is slow — e.g. the current/upcoming leader is unreachable and the QUIC connection setup stalls — the whole send_transaction call is blocked on it, even though the RPC path already succeeded quickly. This is exactly the kind of "blocking call without a timeout on the request path" that can cascade into elevated tail latency/outages for a crate explicitly documented as performance-sensitive ("Changes must avoid increasing confirmation latency" — .agents/context/crates/magicblock-rpc-client.md). Wrap the TPU future in a bounded tokio::time::timeout so a stalled leader connection can't stall the whole send.
🛡️ Proposed fix: bound the TPU send with a timeout
let tpu_send: TpuSendFuture<'_> = Box::pin(async {
let Some(wire_transaction) = wire_transaction else {
warn!(
"Failed to serialize transaction for leader TPU delivery"
);
return false;
};
let Some(tpu_sender) = self.tpu_sender().await else {
return false;
};
- tpu_sender(wire_transaction).await
+ match tokio::time::timeout(
+ TPU_SEND_TIMEOUT,
+ tpu_sender(wire_transaction),
+ )
+ .await
+ {
+ Ok(sent) => sent,
+ Err(_) => {
+ warn!("Leader TPU delivery timed out");
+ false
+ }
+ }
});📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| let wire_transaction = bincode::serialize(tx).ok(); | |
| let rpc_send = self | |
| .client | |
| .send_transaction_with_config(tx, SEND_TRANSACTION_CONFIG) | |
| .await | |
| .map_err(|e| { | |
| MagicBlockRpcClientError::SendTransaction(Box::new(e)) | |
| })?; | |
| .send_transaction_with_config(tx, SEND_TRANSACTION_CONFIG); | |
| // Type-erase this future so every generic transaction caller does not | |
| // inherit the deeply nested TPU client future in its async state machine. | |
| let tpu_send: TpuSendFuture<'_> = Box::pin(async { | |
| let Some(wire_transaction) = wire_transaction else { | |
| warn!( | |
| "Failed to serialize transaction for leader TPU delivery" | |
| ); | |
| return false; | |
| }; | |
| let Some(tpu_sender) = self.tpu_sender().await else { | |
| return false; | |
| }; | |
| tpu_sender(wire_transaction).await | |
| }); | |
| let (rpc_result, tpu_sent) = tokio::join!(rpc_send, tpu_send); | |
| let sig = match rpc_result { | |
| Ok(signature) => signature, | |
| Err(err) if tpu_sent => { | |
| warn!(?err, "RPC transaction delivery failed after leader TPU delivery succeeded"); | |
| *tx.get_signature() | |
| } | |
| Err(err) => { | |
| return Err(MagicBlockRpcClientError::SendTransaction( | |
| Box::new(err), | |
| )); | |
| } | |
| }; | |
| let wire_transaction = bincode::serialize(tx).ok(); | |
| let rpc_send = self | |
| .client | |
| .send_transaction_with_config(tx, SEND_TRANSACTION_CONFIG); | |
| // Type-erase this future so every generic transaction caller does not | |
| // inherit the deeply nested TPU client future in its async state machine. | |
| let tpu_send: TpuSendFuture<'_> = Box::pin(async { | |
| let Some(wire_transaction) = wire_transaction else { | |
| warn!( | |
| "Failed to serialize transaction for leader TPU delivery" | |
| ); | |
| return false; | |
| }; | |
| let Some(tpu_sender) = self.tpu_sender().await else { | |
| return false; | |
| }; | |
| match tokio::time::timeout( | |
| TPU_SEND_TIMEOUT, | |
| tpu_sender(wire_transaction), | |
| ) | |
| .await | |
| { | |
| Ok(sent) => sent, | |
| Err(_) => { | |
| warn!("Leader TPU delivery timed out"); | |
| false | |
| } | |
| } | |
| }); | |
| let (rpc_result, tpu_sent) = tokio::join!(rpc_send, tpu_send); | |
| let sig = match rpc_result { | |
| Ok(signature) => signature, | |
| Err(err) if tpu_sent => { | |
| warn!(?err, "RPC transaction delivery failed after leader TPU delivery succeeded"); | |
| *tx.get_signature() | |
| } | |
| Err(err) => { | |
| return Err(MagicBlockRpcClientError::SendTransaction( | |
| Box::new(err), | |
| )); | |
| } | |
| }; |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@magicblock-rpc-client/src/lib.rs` around lines 696 - 727, Bound the TPU
delivery future in send_transaction with a tokio::time::timeout using the
appropriate existing timeout duration or a clearly defined bounded value. Handle
timeout expiration as an unsuccessful TPU send while preserving the current
serialization, sender lookup, RPC error handling, and signature fallback
behavior; ensure a stalled tpu_sender(wire_transaction) cannot delay completion
after RPC succeeds.
1 participant
Implemented direct leader TPU transaction delivery while keeping the existing RPC load-balancer path as a fallback.
Changes:
Validation:
Part of #749.
Summary by CodeRabbit
New Features
Documentation