= vec![0; 32];
client.send_transaction(&raw_tx).await?;
println!("transaction queued");
Ok(())
}
```
## What To Replace
* Swap `london.nextblock.io:11100` for the region closest to you.
* Replace `raw_tx` with the serialized bytes of your signed transaction.
* Read your API key from `NEXTBLOCK_API_KEY` or your own config loader.
## Notes
* Keep the client alive and reuse it for many sends.
* The server expects raw transaction bytes, not base64.
* QUIC does not support the extra gRPC submission flags or atomic bundle submission.
* If you already build transactions with `solana-sdk`, serializing the signed transaction is enough before calling `send_transaction()`.
See [QUIC Transaction Submission](/api/quic-transaction-submission) for the full endpoint list and protocol summary.
# Submit Single Transaction
Submit individual transactions to NextBlock using Rust and the Solana SDK.
If you want to send raw signed transaction bytes over QUIC instead of gRPC, see [QUIC Transaction Submission](/api/examples/rust/quic).
This example shows transaction construction and the request shape you send to NextBlock. Replace the placeholder generated client types with the client generated from `nextblock-proto`.
## Example
use solana_sdk::{
instruction::Instruction,
message::Message,
pubkey::Pubkey,
signature::{Keypair, Signature, Signer},
system_instruction,
transaction::Transaction,
hash::Hash,
};
use solana_client::rpc_client::RpcClient;
use rand::seq::SliceRandom;
use std::str::FromStr;
// NextBlock tip wallets for load balancing
const NEXTBLOCK_TIP_WALLETS: [&str; 8] = [
"NEXTbLoCkB51HpLBLojQfpyVAMorm3zzKg7w9NFdqid",
"nextBLoCkPMgmG8ZgJtABeScP35qLa2AMCNKntAP7Xc",
"NextbLoCkVtMGcV47JzewQdvBpLqT9TxQFozQkN98pE",
"NexTbLoCkWykbLuB1NkjXgFWkX9oAtcoagQegygXXA2",
"NeXTBLoCKs9F1y5PJS9CKrFNNLU1keHW71rfh7KgA1X",
"NexTBLockJYZ7QD7p2byrUa6df8ndV2WSd8GkbWqfbb",
"neXtBLock1LeC67jYd1QdAa32kbVeubsfPNTJC1V5At",
"nEXTBLockYgngeRmRrjDV31mGSekVPqZoMGhQEZtPVG",
];
// Get random tip wallet for load balancing
fn get_random_nextblock_tip_wallet() -> Result<Pubkey, Box<dyn std::error::Error>> {
let mut rng = rand::thread_rng();
let wallet_str = NEXTBLOCK_TIP_WALLETS.choose(&mut rng)
.ok_or("No tip wallets available")?;
Ok(Pubkey::from_str(wallet_str)?)
}
// Build transaction with tip and instructions
fn build_transaction_with_tip(
payer: &Keypair,
recent_blockhash: Hash,
tip_amount: u64,
instructions: Vec<Instruction>,
) -> Result<Transaction, Box<dyn std::error::Error>> {
let tip_wallet = get_random_nextblock_tip_wallet()?;
// Create tip instruction (should be first)
let tip_instruction = system_instruction::transfer(
&payer.pubkey(),
&tip_wallet,
tip_amount,
);
// Combine tip instruction with user instructions
let mut all_instructions = vec![tip_instruction];
all_instructions.extend(instructions);
// Create and sign transaction
let message = Message::new(&all_instructions, Some(&payer.pubkey()));
let mut transaction = Transaction::new_unsigned(message);
transaction.sign(&[payer], recent_blockhash);
Ok(transaction)
}
// Submit transaction to NextBlock
async fn submit_single_transaction(
// nextblock_client: &mut YourGeneratedApiClient,
rpc_client: &RpcClient,
signer: &Keypair,
recipient: Pubkey,
transfer_amount: u64,
tip_amount: u64,
) -> Result<Signature, Box<dyn std::error::Error>> {
// Get recent blockhash
let recent_blockhash = rpc_client.get_latest_blockhash()?;
// Create transfer instruction
let transfer_instruction = system_instruction::transfer(
&signer.pubkey(),
&recipient,
transfer_amount,
);
// Build transaction with tip
let transaction = build_transaction_with_tip(
signer,
recent_blockhash,
tip_amount,
vec![transfer_instruction],
)?;
// Convert to base64 for submission
let serialized_tx = bincode::serialize(&transaction)?;
let base64_tx = base64::encode(serialized_tx);
// Configure submission options
let front_running_protection = false;
let revert_on_fail = false;
let disable_retries = false;
let snipe_transaction = false;
// Submit to NextBlock
/* Uncomment when you have the generated API client
let request = tonic::Request::new(PostSubmitRequest {
transaction: Some(TransactionMessage {
content: base64_tx,
}),
skip_pre_flight: Some(true),
snipe_transaction: Some(snipe_transaction),
front_running_protection: Some(front_running_protection),
disable_retries: Some(disable_retries),
revert_on_fail: Some(revert_on_fail),
});
let response = nextblock_client.post_submit_v2(request).await?;
let submit_response = response.into_inner();
println!("Transaction submitted successfully!");
println!("Signature: {}", submit_response.signature);
println!("UUID: {}", submit_response.uuid);
Ok(Signature::from_str(&submit_response.signature)?)
*/
println!("Local transaction built successfully: {}", transaction.signatures[0]);
Ok(transaction.signatures[0])
}
// Example with dynamic tip calculation
async fn submit_with_optimal_tip(
// nextblock_client: &mut YourGeneratedApiClient,
rpc_client: &RpcClient,
signer: &Keypair,
recipient: Pubkey,
transfer_amount: u64,
) -> Result<Signature, Box<dyn std::error::Error>> {
// Get optimal tip amount (implement tip floor API call)
let optimal_tip = get_optimal_tip_amount().await?;
submit_single_transaction(
// nextblock_client,
rpc_client,
signer,
recipient,
transfer_amount,
optimal_tip,
).await
}
// Get optimal tip amount from tip floor API
async fn get_optimal_tip_amount() -> Result<u64, Box<dyn std::error::Error>> {
// This should call the tip floor API to get current percentiles
// For now, return a reasonable default
Ok(1_000_000) // 0.001 SOL
}
## Usage Example
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Initialize configuration
let signer = Keypair::new(); // Use your actual keypair
let recipient = Pubkey::from_str("<recipient-public-key>")?;
let rpc_client = RpcClient::new("https://api.mainnet-beta.solana.com".to_string());
// Connect to NextBlock (see connection.md)
// let config = NextBlockConfig::default();
// let mut nextblock_client = create_nextblock_client(config).await?;
// Submit transaction
let signature = submit_single_transaction(
// &mut nextblock_client,
&rpc_client,
&signer,
recipient,
10_000, // Transfer 10,000 lamports
1_000_000, // Tip 1,000,000 lamports (0.001 SOL)
).await?;
println!("Transaction submitted with signature: {}", signature);
// Alternative: Submit with optimal tip
let signature2 = submit_with_optimal_tip(
// &mut nextblock_client,
&rpc_client,
&signer,
recipient,
20_000, // Transfer 20,000 lamports
).await?;
println!("Optimally tipped transaction: {}", signature2);
Ok(())
}
## Advanced Features
### Custom Instruction Building
use solana_sdk::instruction::{AccountMeta, Instruction};
// Build custom program instruction
fn build_custom_instruction(
program_id: Pubkey,
accounts: Vec<AccountMeta>,
data: Vec<u8>,
) -> Instruction {
Instruction {
program_id,
accounts,
data,
}
}
// Example: Token transfer instruction
fn build_token_transfer_instruction(
source: Pubkey,
destination: Pubkey,
authority: Pubkey,
amount: u64,
) -> Instruction {
// This is a simplified example - use spl-token crate for real token transfers
let accounts = vec![
AccountMeta::new(source, false),
AccountMeta::new(destination, false),
AccountMeta::new_readonly(authority, true),
];
build_custom_instruction(
spl_token::id(),
accounts,
amount.to_le_bytes().to_vec(),
)
}
### Error Handling and Retries
use tokio::time::{sleep, Duration};
// Retry logic with exponential backoff
async fn submit_with_retry(
// nextblock_client: &mut YourGeneratedApiClient,
rpc_client: &RpcClient,
signer: &Keypair,
recipient: Pubkey,
transfer_amount: u64,
tip_amount: u64,
max_retries: u32,
) -> Result<Signature, Box<dyn std::error::Error>> {
let mut retry_count = 0;
let mut delay = Duration::from_millis(1000);
loop {
match submit_single_transaction(
// nextblock_client,
rpc_client,
signer,
recipient,
transfer_amount,
tip_amount,
).await {
Ok(signature) => return Ok(signature),
Err(e) if retry_count < max_retries => {
println!("Attempt {} failed: {}. Retrying in {:?}...",
retry_count + 1, e, delay);
sleep(delay).await;
retry_count += 1;
delay *= 2; // Exponential backoff
}
Err(e) => return Err(e),
}
}
}
## Best Practices
1. **Always include tips**: NextBlock prioritizes transactions with appropriate tips
2. **Use random tip wallets**: Distribute load across multiple tip addresses
3. **Monitor tip floors**: Adjust tips based on current network conditions
4. **Handle errors gracefully**: Implement retry logic for network issues
5. **Validate inputs**: Always validate public keys and amounts before submission
6. **Use appropriate RPC endpoints**: Choose reliable RPC providers
7. **Keep connections alive**: Reuse gRPC connections for better performance
# Submit Batched Transactions
Submit 2-4 transactions as an atomic bundle to NextBlock using Rust. Batched transactions are processed as Jito bundles - either all transactions succeed, or none do.
This example shows bundle construction and the request shape you send to NextBlock. Replace the placeholder generated client types with the client generated from `nextblock-proto`.
## Example
use solana_sdk::{
instruction::Instruction,
message::Message,
pubkey::Pubkey,
signature::{Keypair, Signature, Signer},
system_instruction,
transaction::Transaction,
hash::Hash,
};
use solana_client::rpc_client::RpcClient;
use rand::seq::SliceRandom;
use std::str::FromStr;
// NextBlock tip wallets
const NEXTBLOCK_TIP_WALLETS: [&str; 8] = [
"NEXTbLoCkB51HpLBLojQfpyVAMorm3zzKg7w9NFdqid",
"nextBLoCkPMgmG8ZgJtABeScP35qLa2AMCNKntAP7Xc",
"NextbLoCkVtMGcV47JzewQdvBpLqT9TxQFozQkN98pE",
"NexTbLoCkWykbLuB1NkjXgFWkX9oAtcoagQegygXXA2",
"NeXTBLoCKs9F1y5PJS9CKrFNNLU1keHW71rfh7KgA1X",
"NexTBLockJYZ7QD7p2byrUa6df8ndV2WSd8GkbWqfbb",
"neXtBLock1LeC67jYd1QdAa32kbVeubsfPNTJC1V5At",
"nEXTBLockYgngeRmRrjDV31mGSekVPqZoMGhQEZtPVG",
];
// Transaction builder for bundle
struct TransactionBundle {
transactions: Vec<Transaction>,
}
impl TransactionBundle {
fn new() -> Self {
Self {
transactions: Vec::new(),
}
}
// Add transaction to bundle
fn add_transaction(&mut self, transaction: Transaction) -> Self {
self.transactions.push(transaction);
self
}
// Get transactions as base64 strings
fn to_base64_transactions(&self) -> Result<Vec<String>, Box<dyn std::error::Error>> {
self.transactions
.iter()
.map(|tx| {
let serialized = bincode::serialize(tx)?;
Ok(base64::encode(serialized))
})
.collect()
}
}
// Build single transaction with tip
fn build_transaction_with_tip(
payer: &Keypair,
recent_blockhash: Hash,
tip_amount: u64,
instructions: Vec<Instruction>,
) -> Result<Transaction, Box<dyn std::error::Error>> {
let mut rng = rand::thread_rng();
let tip_wallet_str = NEXTBLOCK_TIP_WALLETS.choose(&mut rng)
.ok_or("No tip wallets available")?;
let tip_wallet = Pubkey::from_str(tip_wallet_str)?;
// Create tip instruction (should be first)
let tip_instruction = system_instruction::transfer(
&payer.pubkey(),
&tip_wallet,
tip_amount,
);
// Combine instructions
let mut all_instructions = vec![tip_instruction];
all_instructions.extend(instructions);
// Build and sign transaction
let message = Message::new(&all_instructions, Some(&payer.pubkey()));
let mut transaction = Transaction::new_unsigned(message);
transaction.sign(&[payer], recent_blockhash);
Ok(transaction)
}
// Submit batched transactions
async fn submit_batched_transactions(
// nextblock_client: &mut YourGeneratedApiClient,
rpc_client: &RpcClient,
signer: &Keypair,
) -> Result<String, Box<dyn std::error::Error>> {
// Get recent blockhash (same for all transactions in bundle)
let recent_blockhash = rpc_client.get_latest_blockhash()?;
// Build transaction bundle
let mut bundle = TransactionBundle::new();
// Transaction 1: Setup transaction
let setup_instructions = vec![
system_instruction::transfer(
&signer.pubkey(),
&Pubkey::from_str("<recipient1-public-key>")?,
100_000, // 0.0001 SOL
),
];
let setup_tx = build_transaction_with_tip(
signer,
recent_blockhash,
500_000, // 0.0005 SOL tip
setup_instructions,
)?;
bundle = bundle.add_transaction(setup_tx);
// Transaction 2: Main operation
let main_instructions = vec![
system_instruction::transfer(
&signer.pubkey(),
&Pubkey::from_str("<recipient2-public-key>")?,
200_000, // 0.0002 SOL
),
];
let main_tx = build_transaction_with_tip(
signer,
recent_blockhash,
1_000_000, // 0.001 SOL tip (higher for main transaction)
main_instructions,
)?;
bundle = bundle.add_transaction(main_tx);
// Transaction 3: Cleanup transaction
let cleanup_instructions = vec![
system_instruction::transfer(
&signer.pubkey(),
&Pubkey::from_str("<recipient3-public-key>")?,
50_000, // 0.00005 SOL
),
];
let cleanup_tx = build_transaction_with_tip(
signer,
recent_blockhash,
500_000, // 0.0005 SOL tip
cleanup_instructions,
)?;
bundle = bundle.add_transaction(cleanup_tx);
// Convert to base64 for submission
let base64_transactions = bundle.to_base64_transactions()?;
// Print transaction signatures for tracking
for (i, tx) in bundle.transactions.iter().enumerate() {
println!("Transaction {} signature: {}", i + 1, tx.signatures[0]);
}
// Submit bundle to NextBlock
/* Uncomment when you have the generated API client
let entries: Vec<PostSubmitRequestEntry> = base64_transactions
.into_iter()
.map(|tx_base64| PostSubmitRequestEntry {
transaction: Some(TransactionMessage {
content: tx_base64,
}),
})
.collect();
let request = tonic::Request::new(PostSubmitBatchRequest {
entries,
});
let response = nextblock_client.post_submit_batch_v2(request).await?;
let submit_response = response.into_inner();
println!("Batch submitted successfully!");
println!("Bundle signature: {}", submit_response.signature);
Ok(submit_response.signature)
*/
println!("Local bundle prepared with {} transactions.", bundle.transactions.len());
Ok("".to_string())
}
## Arbitrage Bundle Example
// Example: Build arbitrage bundle
async fn build_arbitrage_bundle(
// nextblock_client: &mut YourGeneratedApiClient,
rpc_client: &RpcClient,
signer: &Keypair,
dex_a_address: Pubkey,
dex_b_address: Pubkey,
trade_amount: u64,
) -> Result<String, Box<dyn std::error::Error>> {
let recent_blockhash = rpc_client.get_latest_blockhash()?;
let mut bundle = TransactionBundle::new();
// Transaction 1: Buy on DEX A
let buy_instructions = vec![
// Add your DEX-specific buy instructions here
system_instruction::transfer(
&signer.pubkey(),
&dex_a_address,
trade_amount,
),
];
let buy_tx = build_transaction_with_tip(
signer,
recent_blockhash,
2_000_000, // Higher tip for arbitrage
buy_instructions,
)?;
bundle = bundle.add_transaction(buy_tx);
// Transaction 2: Sell on DEX B
let sell_instructions = vec![
// Add your DEX-specific sell instructions here
system_instruction::transfer(
&signer.pubkey(),
&dex_b_address,
trade_amount,
),
];
let sell_tx = build_transaction_with_tip(
signer,
recent_blockhash,
2_000_000, // Higher tip for arbitrage
sell_instructions,
)?;
bundle = bundle.add_transaction(sell_tx);
println!("Arbitrage bundle prepared with {} transactions", bundle.transactions.len());
// Submit the bundle
// Implementation similar to submit_batched_transactions
Ok("".to_string())
}
## Complex DeFi Bundle Example
// Example: Complex DeFi operation bundle
async fn build_defi_operation_bundle(
// nextblock_client: &mut YourGeneratedApiClient,
rpc_client: &RpcClient,
signer: &Keypair,
) -> Result<String, Box<dyn std::error::Error>> {
let recent_blockhash = rpc_client.get_latest_blockhash()?;
let mut bundle = TransactionBundle::new();
// Transaction 1: Create token accounts
let account_creation_instructions = vec![
// Add token account creation instructions
system_instruction::create_account(
&signer.pubkey(),
&Keypair::new().pubkey(), // New account
1_000_000, // Rent exemption
165, // Token account space
&spl_token::id(),
),
];
let account_tx = build_transaction_with_tip(
signer,
recent_blockhash,
500_000,
account_creation_instructions,
)?;
bundle = bundle.add_transaction(account_tx);
// Transaction 2: Execute swap
let swap_instructions = vec![
// Add swap instructions (Jupiter, Raydium, etc.)
system_instruction::transfer(
&signer.pubkey(),
&Pubkey::from_str("<swap-program-address>")?,
0, // No SOL transfer for swap
),
];
let swap_tx = build_transaction_with_tip(
signer,
recent_blockhash,
1_500_000, // Higher tip for main operation
swap_instructions,
)?;
bundle = bundle.add_transaction(swap_tx);
// Transaction 3: Stake or provide liquidity
let stake_instructions = vec![
// Add staking/liquidity instructions
system_instruction::transfer(
&signer.pubkey(),
&Pubkey::from_str("<stake-pool-address>")?,
0,
),
];
let stake_tx = build_transaction_with_tip(
signer,
recent_blockhash,
750_000,
stake_instructions,
)?;
bundle = bundle.add_transaction(stake_tx);
println!("DeFi operation bundle prepared with {} transactions", bundle.transactions.len());
Ok("".to_string())
}
## Usage Example
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Initialize configuration
let signer = Keypair::new(); // Use your actual keypair
let rpc_client = RpcClient::new("https://api.mainnet-beta.solana.com".to_string());
// Connect to NextBlock (see connection.md)
// let config = NextBlockConfig::default();
// let mut nextblock_client = create_nextblock_client(config).await?;
// Submit basic batch
let bundle_signature = submit_batched_transactions(
// &mut nextblock_client,
&rpc_client,
&signer,
).await?;
println!("Batch submitted with signature: {}", bundle_signature);
// Submit arbitrage bundle
let arb_signature = build_arbitrage_bundle(
// &mut nextblock_client,
&rpc_client,
&signer,
Pubkey::from_str("<dex-a-address>")?,
Pubkey::from_str("<dex-b-address>")?,
1_000_000, // 0.001 SOL
).await?;
println!("Arbitrage bundle: {}", arb_signature);
// Submit complex DeFi bundle
let defi_signature = build_defi_operation_bundle(
// &mut nextblock_client,
&rpc_client,
&signer,
).await?;
println!("DeFi bundle: {}", defi_signature);
Ok(())
}
## Bundle Best Practices
### Size and Composition
* **Bundle size**: 2-4 transactions (optimal: 2-3)
* **Transaction order**: Setup → Main operations → Cleanup
* **Compute limits**: Stay within bundle compute unit limits
### Tip Strategy
* **Progressive tipping**: Higher tips for more critical transactions
* **Bundle coherence**: All transactions should have meaningful tips
* **Market conditions**: Adjust tips based on network congestion
### Error Handling
// Validate bundle before submission
fn validate_bundle(bundle: &TransactionBundle) -> Result<(), &'static str> {
if bundle.transactions.is_empty() {
return Err("Bundle cannot be empty");
}
if bundle.transactions.len() > 4 {
return Err("Bundle cannot contain more than 4 transactions");
}
if bundle.transactions.len() < 2 {
return Err("Bundle must contain at least 2 transactions");
}
// Check for signature duplicates
let mut signatures = std::collections::HashSet::new();
for tx in &bundle.transactions {
if !signatures.insert(tx.signatures[0]) {
return Err("Duplicate transaction signatures in bundle");
}
}
Ok(())
}
## Common Use Cases
1. **Arbitrage**: Multi-DEX price differences
2. **DeFi operations**: Account setup → Trade → Stake
3. **NFT operations**: Create → Mint → Transfer → List
4. **Token operations**: Create mint → Create accounts → Transfer
5. **Liquidations**: Detect → Execute → Collect rewards
## Bundle Failure Prevention
* **Account validation**: Ensure all accounts exist and have sufficient balance
* **Instruction compatibility**: Avoid conflicting program interactions
* **Blockhash freshness**: Use recent blockhashes (valid \~60 seconds)
* **Compute budget**: Monitor total compute units across all transactions
# Tip Floor Stream
Stream real-time tip floor data from NextBlock to optimize your transaction tips dynamically.
## Streaming Tip Floor Data
use tokio_stream::StreamExt;
use std::time::Duration;
use serde::{Deserialize, Serialize};
// Tip floor response structure
#[derive(Debug, Deserialize, Serialize)]
pub struct TipFloorData {
pub time: String,
pub landed_tips_25th_percentile: f64,
pub landed_tips_50th_percentile: f64,
pub landed_tips_75th_percentile: f64,
pub landed_tips_95th_percentile: f64,
pub landed_tips_99th_percentile: f64,
pub ema_landed_tips_50th_percentile: f64,
}
// Stream tip floor updates
async fn stream_tip_floor(
// nextblock_client: &mut YourGeneratedApiClient,
update_frequency: String,
) -> Result<(), Box<dyn std::error::Error>> {
println!("Starting tip floor stream with frequency: {}", update_frequency);
// Create streaming request
/* Uncomment when you have the generated API client
let request = tonic::Request::new(TipFloorStreamRequest {
update_frequency,
});
let mut stream = nextblock_client
.stream_tip_floor(request)
.await?
.into_inner();
println!("Streaming tip floor data:");
while let Some(tip_floor_response) = stream.next().await {
match tip_floor_response {
Ok(tip_floor) => {
println!("Received tip floor update:");
println!(" Time: {}", tip_floor.time);
println!(" 25th percentile: {:.6} SOL", tip_floor.landed_tips_25th_percentile);
println!(" 50th percentile: {:.6} SOL", tip_floor.landed_tips_50th_percentile);
println!(" 75th percentile: {:.6} SOL", tip_floor.landed_tips_75th_percentile);
println!(" 95th percentile: {:.6} SOL", tip_floor.landed_tips_95th_percentile);
println!(" 99th percentile: {:.6} SOL", tip_floor.landed_tips_99th_percentile);
println!(" EMA 50th percentile: {:.6} SOL", tip_floor.ema_landed_tips_50th_percentile);
println!(" ---");
// Process the tip floor data
process_tip_floor_update(&tip_floor).await?;
}
Err(e) => {
eprintln!("Stream error: {}", e);
// Implement reconnection logic here
break;
}
}
}
*/
// Mock streaming for demonstration
println!("Mock tip floor streaming started...");
let mut interval = tokio::time::interval(Duration::from_secs(60));
loop {
interval.tick().await;
let mock_tip_floor = TipFloorData {
time: chrono::Utc::now().to_rfc3339(),
landed_tips_25th_percentile: 0.0011,
landed_tips_50th_percentile: 0.005000001,
landed_tips_75th_percentile: 0.01555,
landed_tips_95th_percentile: 0.09339195639999975,
landed_tips_99th_percentile: 0.4846427910400001,
ema_landed_tips_50th_percentile: 0.005989477267191758,
};
println!("Mock tip floor update: {:#?}", mock_tip_floor);
process_tip_floor_update(&mock_tip_floor).await?;
}
}
// Process tip floor updates
async fn process_tip_floor_update(
tip_floor: &TipFloorData,
) -> Result<(), Box<dyn std::error::Error>> {
// Update global tip strategy
update_tip_strategy(tip_floor).await?;
// Optionally store historical data
store_tip_floor_data(tip_floor).await?;
// Trigger any pending transactions with updated tips
trigger_pending_transactions().await?;
Ok(())
}
## Dynamic Tip Calculation
use std::sync::Arc;
use tokio::sync::RwLock;
// Global tip strategy state
#[derive(Debug, Clone)]
pub struct TipStrategy {
pub conservative_tip: u64, // 25th percentile
pub normal_tip: u64, // 50th percentile
pub aggressive_tip: u64, // 75th percentile
pub priority_tip: u64, // 95th percentile
pub last_updated: chrono::DateTime<chrono::Utc>,
}
impl Default for TipStrategy {
fn default() -> Self {
Self {
conservative_tip: 500_000, // 0.0005 SOL
normal_tip: 1_000_000, // 0.001 SOL
aggressive_tip: 2_000_000, // 0.002 SOL
priority_tip: 5_000_000, // 0.005 SOL
last_updated: chrono::Utc::now(),
}
}
}
// Global tip strategy instance
static TIP_STRATEGY: once_cell::sync::Lazy<Arc<RwLock<TipStrategy>>> =
once_cell::sync::Lazy::new(|| Arc::new(RwLock::new(TipStrategy::default())));
// Convert SOL to lamports
fn sol_to_lamports(sol: f64) -> u64 {
(sol * 1_000_000_000.0) as u64
}
// Update tip strategy based on tip floor data
async fn update_tip_strategy(
tip_floor: &TipFloorData,
) -> Result<(), Box<dyn std::error::Error>> {
let mut strategy = TIP_STRATEGY.write().await;
strategy.conservative_tip = sol_to_lamports(tip_floor.landed_tips_25th_percentile);
strategy.normal_tip = sol_to_lamports(tip_floor.landed_tips_50th_percentile);
strategy.aggressive_tip = sol_to_lamports(tip_floor.landed_tips_75th_percentile);
strategy.priority_tip = sol_to_lamports(tip_floor.landed_tips_95th_percentile);
strategy.last_updated = chrono::Utc::now();
println!("Updated tip strategy:");
println!(" Conservative: {} lamports ({:.6} SOL)",
strategy.conservative_tip, tip_floor.landed_tips_25th_percentile);
println!(" Normal: {} lamports ({:.6} SOL)",
strategy.normal_tip, tip_floor.landed_tips_50th_percentile);
println!(" Aggressive: {} lamports ({:.6} SOL)",
strategy.aggressive_tip, tip_floor.landed_tips_75th_percentile);
println!(" Priority: {} lamports ({:.6} SOL)",
strategy.priority_tip, tip_floor.landed_tips_95th_percentile);
Ok(())
}
// Get optimal tip for transaction priority
pub async fn get_optimal_tip(priority: TipPriority) -> u64 {
let strategy = TIP_STRATEGY.read().await;
match priority {
TipPriority::Conservative => strategy.conservative_tip,
TipPriority::Normal => strategy.normal_tip,
TipPriority::Aggressive => strategy.aggressive_tip,
TipPriority::Priority => strategy.priority_tip,
}
}
// Tip priority levels
#[derive(Debug, Clone, Copy)]
pub enum TipPriority {
Conservative, // 25th percentile - cheapest option
Normal, // 50th percentile - balanced option
Aggressive, // 75th percentile - faster execution
Priority, // 95th percentile - highest priority
}
## Advanced Tip Management
use std::collections::VecDeque;
// Historical tip data for trend analysis
#[derive(Debug)]
pub struct TipHistory {
data: VecDeque<TipFloorData>,
max_size: usize,
}
impl TipHistory {
pub fn new(max_size: usize) -> Self {
Self {
data: VecDeque::with_capacity(max_size),
max_size,
}
}
pub fn add(&mut self, tip_floor: TipFloorData) {
if self.data.len() >= self.max_size {
self.data.pop_front();
}
self.data.push_back(tip_floor);
}
// Calculate trend (increasing/decreasing tips)
pub fn calculate_trend(&self) -> f64 {
if self.data.len() < 2 {
return 0.0;
}
let recent = &self.data[self.data.len() - 1];
let older = &self.data[self.data.len() - 2];
recent.landed_tips_50th_percentile - older.landed_tips_50th_percentile
}
// Get average tip over time window
pub fn get_average_tip(&self, percentile: &str) -> f64 {
if self.data.is_empty() {
return 0.0;
}
let sum: f64 = self.data.iter().map(|d| {
match percentile {
"25th" => d.landed_tips_25th_percentile,
"50th" => d.landed_tips_50th_percentile,
"75th" => d.landed_tips_75th_percentile,
"95th" => d.landed_tips_95th_percentile,
_ => d.landed_tips_50th_percentile,
}
}).sum();
sum / self.data.len() as f64
}
}
// Smart tip calculation with trend analysis
pub async fn get_smart_tip(
base_priority: TipPriority,
tip_history: &TipHistory,
) -> u64 {
let base_tip = get_optimal_tip(base_priority).await;
let trend = tip_history.calculate_trend();
// Adjust tip based on trend
let adjustment_factor = if trend > 0.001 {
1.2 // Tips are increasing, be more aggressive
} else if trend < -0.001 {
0.9 // Tips are decreasing, can be more conservative
} else {
1.0 // No significant trend
};
((base_tip as f64) * adjustment_factor) as u64
}
## Store Historical Data
// Store tip floor data for analysis
async fn store_tip_floor_data(
tip_floor: &TipFloorData,
) -> Result<(), Box<dyn std::error::Error>> {
// Example: Store to file (in production, use a database)
let data_dir = std::path::Path::new("./tip_data");
if !data_dir.exists() {
tokio::fs::create_dir_all(data_dir).await?;
}
let filename = format!("tip_floor_{}.json",
chrono::Utc::now().format("%Y%m%d"));
let filepath = data_dir.join(filename);
// Append to daily file
let json_line = format!("{}\n", serde_json::to_string(tip_floor)?);
tokio::fs::write(&filepath, json_line).await?;
println!("Stored tip floor data to {:?}", filepath);
Ok(())
}
// Trigger pending transactions with updated tips
async fn trigger_pending_transactions() -> Result<(), Box<dyn std::error::Error>> {
// This would check for any pending transactions that are waiting
// for better tip conditions and submit them with updated tips
println!("Checking for pending transactions to trigger...");
// Example: Check if any transactions are waiting for lower tips
// and submit them now if conditions are favorable
Ok(())
}
## Usage Example
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Connect to NextBlock (see connection.md)
// let config = NextBlockConfig::default();
// let mut nextblock_client = create_nextblock_client(config).await?;
// Start tip floor streaming in background
let stream_handle = tokio::spawn(async move {
if let Err(e) = stream_tip_floor(
// &mut nextblock_client,
"1m".to_string(), // Update every minute
).await {
eprintln!("Tip floor streaming error: {}", e);
}
});
// Example: Use dynamic tips in transaction submission
tokio::time::sleep(Duration::from_secs(5)).await; // Wait for initial data
let conservative_tip = get_optimal_tip(TipPriority::Conservative).await;
let normal_tip = get_optimal_tip(TipPriority::Normal).await;
let aggressive_tip = get_optimal_tip(TipPriority::Aggressive).await;
println!("Current optimal tips:");
println!(" Conservative: {} lamports", conservative_tip);
println!(" Normal: {} lamports", normal_tip);
println!(" Aggressive: {} lamports", aggressive_tip);
// Use these tips in your transaction submissions
// submit_transaction_with_tip(normal_tip).await?;
// Keep the stream running
stream_handle.await??;
Ok(())
}
## Best Practices
1. **Update frequency**: Use 1-5 minute intervals for most applications
2. **Trend analysis**: Consider tip trends when calculating optimal amounts
3. **Fallback values**: Always have default tip values in case streaming fails
4. **Historical data**: Store tip floor data for analysis and optimization
5. **Priority levels**: Use different tip strategies for different transaction types
6. **Error handling**: Implement reconnection logic for stream interruptions
7. **Resource management**: Limit historical data storage to prevent memory issues
# Keepalive
Maintain persistent gRPC connections to NextBlock for optimal performance using Rust's Tonic client.
## Basic Keepalive Implementation
use std::time::Duration;
use tokio::time::{interval, sleep};
use tonic::Request;
// Send periodic ping to keep connection alive
async fn start_keepalive_task(
// mut nextblock_client: YourGeneratedApiClient,
ping_interval: Duration,
) -> Result<(), Box<dyn std::error::Error>> {
let mut interval_timer = interval(ping_interval);
println!("Starting keepalive task with interval: {:?}", ping_interval);
loop {
interval_timer.tick().await;
// Send ping request
/* Uncomment when you have the generated API client
match nextblock_client.ping(Request::new(())).await {
Ok(_) => {
println!("Keepalive ping successful at {}",
chrono::Utc::now().format("%H:%M:%S"));
}
Err(e) => {
eprintln!("Keepalive ping failed: {}", e);
// Optionally implement reconnection logic here
break;
}
}
*/
// Mock ping for demonstration
println!("Mock keepalive ping sent at {}",
chrono::Utc::now().format("%H:%M:%S"));
}
Ok(())
}
## Advanced Keepalive with Connection Management
use std::sync::Arc;
use tokio::sync::RwLock;
use tonic::transport::Channel;
// Connection health tracker
#[derive(Debug, Clone)]
pub struct ConnectionHealth {
pub is_healthy: bool,
pub last_successful_ping: chrono::DateTime<chrono::Utc>,
pub consecutive_failures: u32,
pub total_pings_sent: u64,
pub total_pings_successful: u64,
}
impl Default for ConnectionHealth {
fn default() -> Self {
Self {
is_healthy: true,
last_successful_ping: chrono::Utc::now(),
consecutive_failures: 0,
total_pings_sent: 0,
total_pings_successful: 0,
}
}
}
// Advanced keepalive manager
pub struct KeepaliveManager {
// client: Arc<RwLock<YourGeneratedApiClient>>,
health: Arc<RwLock<ConnectionHealth>>,
config: KeepaliveConfig,
}
#[derive(Debug, Clone)]
pub struct KeepaliveConfig {
pub ping_interval: Duration,
pub max_consecutive_failures: u32,
pub reconnect_delay: Duration,
pub health_check_enabled: bool,
}
impl Default for KeepaliveConfig {
fn default() -> Self {
Self {
ping_interval: Duration::from_secs(60),
max_consecutive_failures: 3,
reconnect_delay: Duration::from_secs(5),
health_check_enabled: true,
}
}
}
impl KeepaliveManager {
pub fn new(
// client: YourGeneratedApiClient,
config: KeepaliveConfig,
) -> Self {
Self {
// client: Arc::new(RwLock::new(client)),
health: Arc::new(RwLock::new(ConnectionHealth::default())),
config,
}
}
// Start keepalive task
pub async fn start(&self) -> Result<(), Box<dyn std::error::Error>> {
let health = Arc::clone(&self.health);
// let client = Arc::clone(&self.client);
let config = self.config.clone();
tokio::spawn(async move {
let mut interval_timer = interval(config.ping_interval);
loop {
interval_timer.tick().await;
// Send ping and update health
Self::send_ping_and_update_health(
// Arc::clone(&client),
Arc::clone(&health),
&config,
).await;
}
});
Ok(())
}
// Send ping and update connection health
async fn send_ping_and_update_health(
// client: Arc<RwLock<YourGeneratedApiClient>>,
health: Arc<RwLock<ConnectionHealth>>,
config: &KeepaliveConfig,
) {
let ping_start = std::time::Instant::now();
// Update ping attempt counter
{
let mut health_guard = health.write().await;
health_guard.total_pings_sent += 1;
}
// Send ping
/* Uncomment when you have the generated API client
let ping_result = {
let mut client_guard = client.write().await;
client_guard.ping(Request::new(())).await
};
match ping_result {
Ok(_) => {
let ping_duration = ping_start.elapsed();
let mut health_guard = health.write().await;
health_guard.is_healthy = true;
health_guard.last_successful_ping = chrono::Utc::now();
health_guard.consecutive_failures = 0;
health_guard.total_pings_successful += 1;
println!("Keepalive ping successful ({}ms) - Health: {:.1}%",
ping_duration.as_millis(),
(health_guard.total_pings_successful as f64 /
health_guard.total_pings_sent as f64) * 100.0);
}
Err(e) => {
let mut health_guard = health.write().await;
health_guard.consecutive_failures += 1;
if health_guard.consecutive_failures >= config.max_consecutive_failures {
health_guard.is_healthy = false;
eprintln!("Connection marked as unhealthy after {} consecutive failures",
health_guard.consecutive_failures);
}
eprintln!("Keepalive ping failed (attempt {}): {}",
health_guard.consecutive_failures, e);
}
}
*/
// Mock ping result for demonstration
let ping_duration = ping_start.elapsed();
let mut health_guard = health.write().await;
health_guard.is_healthy = true;
health_guard.last_successful_ping = chrono::Utc::now();
health_guard.consecutive_failures = 0;
health_guard.total_pings_successful += 1;
println!("Mock keepalive ping successful ({}ms) - Health: {:.1}%",
ping_duration.as_millis(),
(health_guard.total_pings_successful as f64 /
health_guard.total_pings_sent as f64) * 100.0);
}
// Get connection health status
pub async fn get_health(&self) -> ConnectionHealth {
self.health.read().await.clone()
}
// Check if connection is healthy
pub async fn is_healthy(&self) -> bool {
self.health.read().await.is_healthy
}
}
## Connection Recovery
// Automatic connection recovery
pub struct ConnectionManager {
config: NextBlockConfig,
keepalive_manager: Option<KeepaliveManager>,
// client: Option<YourGeneratedApiClient>,
}
impl ConnectionManager {
pub fn new(config: NextBlockConfig) -> Self {
Self {
config,
keepalive_manager: None,
// client: None,
}
}
// Establish connection with automatic recovery
pub async fn connect_with_recovery(&mut self) -> Result<(), Box<dyn std::error::Error>> {
loop {
match self.attempt_connection().await {
Ok(()) => {
println!("Successfully connected to NextBlock");
// Start keepalive
self.start_keepalive().await?;
// Monitor connection health
self.monitor_connection_health().await?;
break;
}
Err(e) => {
eprintln!("Connection failed: {}. Retrying in 5 seconds...", e);
sleep(Duration::from_secs(5)).await;
}
}
}
Ok(())
}
// Attempt to establish connection
async fn attempt_connection(&mut self) -> Result<(), Box<dyn std::error::Error>> {
// let client = create_nextblock_client(self.config.clone()).await?;
// self.client = Some(client);
println!("Mock connection established");
Ok(())
}
// Start keepalive task
async fn start_keepalive(&mut self) -> Result<(), Box<dyn std::error::Error>> {
let keepalive_config = KeepaliveConfig::default();
// let keepalive_manager = KeepaliveManager::new(
// self.client.as_ref().unwrap().clone(),
// keepalive_config,
// );
// keepalive_manager.start().await?;
// self.keepalive_manager = Some(keepalive_manager);
println!("Keepalive task started");
Ok(())
}
// Monitor connection health and trigger recovery
async fn monitor_connection_health(&self) -> Result<(), Box<dyn std::error::Error>> {
let mut health_check_interval = interval(Duration::from_secs(30));
loop {
health_check_interval.tick().await;
if let Some(ref keepalive_manager) = self.keepalive_manager {
if !keepalive_manager.is_healthy().await {
eprintln!("Connection unhealthy, attempting recovery...");
// Trigger reconnection logic
return Err("Connection lost".into());
}
}
}
}
}
## Usage Example
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Basic keepalive usage
// let config = NextBlockConfig::default();
// let client = create_nextblock_client(config).await?;
// Start simple keepalive task
let keepalive_handle = tokio::spawn(async move {
start_keepalive_task(
// client,
Duration::from_secs(60), // Ping every minute
).await
});
// Advanced usage with connection management
let config = NextBlockConfig::default();
let connection_manager = ConnectionManager::new(config);
let manager_handle = tokio::spawn(async move {
if let Err(e) = connection_manager.connect_with_recovery().await {
eprintln!("Connection manager failed: {}", e);
}
});
// Your main application logic here
println!("Application running with keepalive...");
// Wait for tasks
tokio::select! {
result = keepalive_handle => {
if let Err(e) = result? {
eprintln!("Keepalive task failed: {}", e);
}
}
result = manager_handle => {
if let Err(e) = result {
eprintln!("Connection manager task failed: {}", e);
}
}
}
Ok(())
}
## Best Practices
1. **Ping frequency**: Use 60-second intervals for most applications
2. **Health monitoring**: Track connection health and implement recovery
3. **Error handling**: Handle ping failures gracefully with exponential backoff
4. **Resource cleanup**: Properly close connections on shutdown
5. **Logging**: Log keepalive status for debugging and monitoring
6. **Timeouts**: Set appropriate timeouts for ping requests
7. **Reconnection logic**: Implement automatic reconnection on connection loss
# Python
Complete Python examples for integrating with NextBlock's gRPC API and QUIC transaction submission using grpcio and solders/solana-py.
## Overview
These examples demonstrate how to:
* Establish secure gRPC connections with authentication
* Submit raw transaction bytes over QUIC for lower submission overhead
* Submit single and batched transactions with proper tipping
* Stream real-time tip floor data for dynamic tip optimization
* Maintain persistent connections with keepalive mechanisms
## Prerequisites
Install the required dependencies:
```bash
pip install grpcio grpcio-tools
pip install solders solana
pip install asyncio aiogrpc
pip install base58 base64
pip install requests
# Generate Python gRPC client from nextblock-proto
# See https://github.com/nextblock-ag/nextblock-proto for instructions
```
## Examples
### Core Examples
* [Connection](/api/examples/python/connection) - Establish gRPC connections with authentication
* [QUIC Transaction Submission](/api/examples/python/quic) - Send raw signed transaction bytes over QUIC
* [Submit Single Transaction](/api/examples/python/submit-single-transactions) - Send individual transactions with tips
* [Submit Batched Transactions](/api/examples/python/submit-batched-transactions) - Send atomic transaction bundles
* [Tip Floor Stream](/api/examples/python/tip-floor-stream) - Stream real-time tip floor data
* [Keepalive](/api/examples/python/keepalive) - Maintain persistent connections
## Quick Start
1. **Generate gRPC client** from [nextblock-proto](https://github.com/nextblock-ag/nextblock-proto)
2. **Install dependencies** using pip
3. **Set up environment** with API key and endpoint
4. **Start with connection example** to establish authenticated gRPC connection
5. **Use tip floor streaming** to optimize transaction tips dynamically
6. **Submit transactions** using single or batched submission methods
## Key Features
### Async/Await Support
All examples use Python's asyncio for non-blocking operations and better performance.
### Type Hints
Complete type annotations for better code clarity and IDE support.
### Error Handling
Comprehensive error handling with retry logic and exponential backoff.
### Connection Management
Persistent connections with keepalive, health monitoring, and automatic recovery.
### Dynamic Tipping
Real-time tip optimization based on current network conditions from tip floor API.
## Best Practices
1. **Use async/await** - Leverage Python's asyncio for better performance
2. **Enable TLS** - Always use secure connections in production
3. **Implement keepalive** - Maintain persistent connections
4. **Monitor tip floors** - Use streaming API for dynamic tip adjustment
5. **Handle errors gracefully** - Implement proper retry logic
6. **Use type hints** - Improve code clarity and maintainability
7. **Choose appropriate endpoints** - Use the closest endpoint for better latency
# Connection
Establish a gRPC connection to NextBlock's API using Python and grpcio.
This page shows the connection pattern and authentication flow. Replace the placeholder client creation with the client generated from [`nextblock-proto`](https://github.com/nextblock-ag/nextblock-proto).
## Prerequisites
Install the required dependencies:
```bash
pip install grpcio grpcio-tools
pip install solders solana
pip install asyncio
```
Generate the Python gRPC client from the proto specs:
```bash
# Clone the proto repository
git clone https://github.com/nextblock-ag/nextblock-proto
cd nextblock-proto
# Generate Python gRPC client
python -m grpc_tools.protoc --python_out=. --grpc_python_out=. -I. *.proto
```
## Connection Setup
import asyncio
import grpc
from typing import Optional, Dict, Any
from dataclasses import dataclass
import ssl
# Authentication metadata interceptor
class AuthInterceptor(grpc.aio.ClientInterceptor):
def __init__(self, api_key: str):
self.api_key = api_key
async def intercept_unary_unary(self, continuation, client_call_details, request):
# Add authorization header to all requests
metadata = list(client_call_details.metadata or [])
metadata.append(('authorization', self.api_key))
new_details = client_call_details._replace(metadata=metadata)
return await continuation(new_details, request)
async def intercept_unary_stream(self, continuation, client_call_details, request):
# Add authorization header to streaming requests
metadata = list(client_call_details.metadata or [])
metadata.append(('authorization', self.api_key))
new_details = client_call_details._replace(metadata=metadata)
return await continuation(new_details, request)
# Connection configuration
@dataclass
class NextBlockConfig:
endpoint: str = "frankfurt.nextblock.io:443"
api_key: str = ""
use_tls: bool = True
timeout: float = 30.0
keepalive_time_ms: int = 60000 # 60 seconds
keepalive_timeout_ms: int = 15000 # 15 seconds
@classmethod
def from_env(cls) -> 'NextBlockConfig':
import os
return cls(
endpoint=os.getenv('NEXTBLOCK_ENDPOINT', 'frankfurt.nextblock.io:443'),
api_key=os.getenv('NEXTBLOCK_API_KEY', ''),
use_tls=os.getenv('NEXTBLOCK_USE_TLS', 'true').lower() == 'true',
timeout=float(os.getenv('NEXTBLOCK_TIMEOUT', '30')),
)
# Create secure channel with authentication
async def create_nextblock_channel(config: NextBlockConfig) -> grpc.aio.Channel:
# Configure channel options
options = [
('grpc.keepalive_time_ms', config.keepalive_time_ms),
('grpc.keepalive_timeout_ms', config.keepalive_timeout_ms),
('grpc.keepalive_permit_without_stream', True),
('grpc.http2.max_pings_without_data', 0),
('grpc.http2.min_ping_interval_without_data_ms', 300000), # 5 minutes
]
if config.use_tls:
# Create secure channel with TLS
credentials = grpc.ssl_channel_credentials()
channel = grpc.aio.secure_channel(
config.endpoint,
credentials,
options=options
)
else:
# Create insecure channel
channel = grpc.aio.insecure_channel(
config.endpoint,
options=options
)
return channel
# Create authenticated client
async def create_nextblock_client(config: NextBlockConfig):
# Create the channel
channel = await create_nextblock_channel(config)
# Add authentication interceptor
auth_interceptor = AuthInterceptor(config.api_key)
intercept_channel = grpc.aio.intercept_channel(channel, auth_interceptor)
# Create the client stub
# Import your generated gRPC client here
# from your_generated_proto import api_pb2_grpc
# client = api_pb2_grpc.ApiStub(intercept_channel)
return intercept_channel, None # Replace None with your generated client stub
# Connection manager with health checking
class NextBlockConnectionManager:
def __init__(self, config: NextBlockConfig):
self.config = config
self.channel: Optional[grpc.aio.Channel] = None
self.client = None
self.is_connected = False
async def connect(self) -> bool:
"""Establish connection to NextBlock"""
try:
self.channel, self.client = await create_nextblock_client(self.config)
# Test the connection
await self.health_check()
self.is_connected = True
print(f"Successfully connected to NextBlock at {self.config.endpoint}")
return True
except Exception as e:
print(f"Failed to connect to NextBlock: {e}")
self.is_connected = False
return False
async def health_check(self) -> bool:
"""Check if the connection is healthy"""
if not self.channel:
return False
try:
# Test connection with a simple call
# await self.client.ping(Empty()) # Uncomment with real client
print("Connection health check passed")
return True
except Exception as e:
print(f"Health check failed: {e}")
return False
async def disconnect(self):
"""Close the connection"""
if self.channel:
await self.channel.close()
self.is_connected = False
print("Disconnected from NextBlock")
async def __aenter__(self):
await self.connect()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.disconnect()
## Usage Example
async def main():
# Configure connection
config = NextBlockConfig(
endpoint="frankfurt.nextblock.io:443",
api_key="<your-api-key-here>",
use_tls=True,
timeout=30.0
)
# Method 1: Direct connection
channel, client = await create_nextblock_client(config)
try:
# Use the client for API calls
print("Connected to NextBlock!")
# Your API calls would go here
# response = await client.ping(Empty())
finally:
await channel.close()
# Method 2: Using connection manager (recommended)
async with NextBlockConnectionManager(config) as manager:
if manager.is_connected:
print("Connection manager established connection")
# Use manager.client for API calls
# response = await manager.client.ping(Empty())
else:
print("Failed to establish connection")
# Run the example
if __name__ == "__main__":
asyncio.run(main())
## Advanced Connection Features
import logging
from typing import Callable, Any
import backoff
# Enhanced connection manager with retry logic
class EnhancedConnectionManager(NextBlockConnectionManager):
def __init__(self, config: NextBlockConfig, max_retries: int = 3):
super().__init__(config)
self.max_retries = max_retries
self.retry_count = 0
@backoff.on_exception(
backoff.expo,
(grpc.RpcError, ConnectionError),
max_tries=3,
base=2,
max_value=60
)
async def connect_with_retry(self) -> bool:
"""Connect with exponential backoff retry"""
self.retry_count += 1
logging.info(f"Connection attempt {self.retry_count}")
success = await self.connect()
if not success:
raise ConnectionError(f"Failed to connect on attempt {self.retry_count}")
self.retry_count = 0 # Reset on success
return True
async def call_with_retry(self, func: Callable, *args, **kwargs) -> Any:
"""Execute gRPC call with automatic retry"""
@backoff.on_exception(
backoff.expo,
grpc.RpcError,
max_tries=3,
giveup=lambda e: e.code() == grpc.StatusCode.UNAUTHENTICATED
)
async def _call():
return await func(*args, **kwargs)
return await _call()
# Connection pool for high-throughput applications
class ConnectionPool:
def __init__(self, config: NextBlockConfig, pool_size: int = 5):
self.config = config
self.pool_size = pool_size
self.connections: List[NextBlockConnectionManager] = []
self.current_index = 0
async def initialize(self):
"""Initialize connection pool"""
for i in range(self.pool_size):
manager = NextBlockConnectionManager(self.config)
if await manager.connect():
self.connections.append(manager)
else:
logging.warning(f"Failed to create connection {i+1}/{self.pool_size}")
logging.info(f"Connection pool initialized with {len(self.connections)} connections")
def get_connection(self) -> NextBlockConnectionManager:
"""Get next available connection (round-robin)"""
if not self.connections:
raise RuntimeError("No available connections in pool")
connection = self.connections[self.current_index]
self.current_index = (self.current_index + 1) % len(self.connections)
return connection
async def close_all(self):
"""Close all connections in the pool"""
for connection in self.connections:
await connection.disconnect()
self.connections.clear()
## Environment Configuration
import os
from typing import Dict, Any
# Load configuration from environment variables
def load_config_from_env() -> NextBlockConfig:
return NextBlockConfig(
endpoint=os.getenv('NEXTBLOCK_ENDPOINT', 'frankfurt.nextblock.io:443'),
api_key=os.getenv('NEXTBLOCK_API_KEY', ''),
use_tls=os.getenv('NEXTBLOCK_USE_TLS', 'true').lower() == 'true',
timeout=float(os.getenv('NEXTBLOCK_TIMEOUT', '30.0')),
keepalive_time_ms=int(os.getenv('NEXTBLOCK_KEEPALIVE_TIME_MS', '60000')),
keepalive_timeout_ms=int(os.getenv('NEXTBLOCK_KEEPALIVE_TIMEOUT_MS', '15000')),
)
# Configuration validation
def validate_config(config: NextBlockConfig) -> bool:
if not config.api_key:
raise ValueError("API key is required")
if not config.endpoint:
raise ValueError("Endpoint is required")
if config.timeout <= 0:
raise ValueError("Timeout must be positive")
return True
# Example with configuration validation
async def main_with_validation():
try:
config = load_config_from_env()
validate_config(config)
async with EnhancedConnectionManager(config) as manager:
await manager.connect_with_retry()
print("Successfully connected with retry logic!")
except ValueError as e:
print(f"Configuration error: {e}")
except Exception as e:
print(f"Connection error: {e}")
if __name__ == "__main__":
asyncio.run(main_with_validation())
## Available Endpoints
* **Frankfurt**: `frankfurt.nextblock.io` (Europe)
* **Amsterdam**: `amsterdam.nextblock.io` (Europe)
* **London**: `london.nextblock.io` (Europe)
* **Singapore**: `singapore.nextblock.io` (Asia)
* **Tokyo**: `tokyo.nextblock.io` (Asia)
* **New York**: `ny.nextblock.io` (US East)
* **Salt Lake City**: `slc.nextblock.io` (US West)
* **Dublin**: `dublin.nextblock.io` (Europe)
* **Vilnius**: `vilnius.nextblock.io` (Europe)
## Best Practices
1. **Use async/await**: Leverage Python's asyncio for non-blocking operations
2. **Use TLS by default**: Prefer secure connections unless you explicitly operate in a trusted internal environment
3. **Implement keepalive**: Configure appropriate keepalive settings
4. **Handle errors gracefully**: Use retry logic with exponential backoff
5. **Validate configuration**: Check all required settings before connecting
6. **Use connection pooling**: For high-throughput applications
7. **Monitor connection health**: Implement regular health checks
8. **Close connections properly**: Always close connections when done
# QUIC Transaction Submission
Use QUIC when you already have signed Solana transaction bytes and want a low-overhead submission path from Python.
The client below:
1. Connects to a regional QUIC endpoint
2. Authenticates with your API key on a bidirectional stream
3. Reuses the connection to send raw transaction bytes on unidirectional streams
## Example
```python
import asyncio
import os
import ssl
import certifi
from aioquic.asyncio.client import connect
from aioquic.quic.configuration import QuicConfiguration
AUTH_OK = b"\x00"
MAX_TX_SIZE = 1232
class NextblockQuicClient:
def __init__(self, client_cm, protocol):
self._client_cm = client_cm
self._protocol = protocol
@classmethod
async def connect(cls, host: str, port: int, api_key: str) -> "NextblockQuicClient":
configuration = QuicConfiguration(
is_client=True,
alpn_protocols=["nb-tx/1"],
)
configuration.verify_mode = ssl.CERT_REQUIRED
configuration.load_verify_locations(cafile=certifi.where())
client_cm = connect(
host,
port,
configuration=configuration,
server_name=host,
)
protocol = await client_cm.__aenter__()
reader, writer = await protocol.create_stream()
writer.write(api_key.encode("utf-8"))
await writer.drain()
writer.write_eof()
response = await reader.readexactly(1)
if response != AUTH_OK:
await client_cm.__aexit__(None, None, None)
raise RuntimeError("authentication rejected")
return cls(client_cm, protocol)
async def send_transaction(self, raw_tx: bytes) -> None:
if len(raw_tx) > MAX_TX_SIZE:
raise ValueError(f"transaction too large: {len(raw_tx)}")
_, writer = await self._protocol.create_stream(is_unidirectional=True)
writer.write(raw_tx)
await writer.drain()
writer.write_eof()
async def close(self) -> None:
await self._client_cm.__aexit__(None, None, None)
async def main() -> None:
api_key = os.environ.get("NEXTBLOCK_API_KEY")
if not api_key:
raise RuntimeError("Set NEXTBLOCK_API_KEY before running")
client = await NextblockQuicClient.connect(
"london.nextblock.io",
11100,
api_key,
)
try:
# Replace this with the serialized bytes of your signed transaction.
# Example with solders: raw_tx = bytes(signed_transaction)
raw_tx = b"\x00\x01\x02\x03"
await client.send_transaction(raw_tx)
print("transaction queued")
finally:
await client.close()
asyncio.run(main())
```
## What To Replace
* Read the API key from `NEXTBLOCK_API_KEY` or your own config source.
* Pick the regional host closest to your deployment.
* Replace `raw_tx` with the serialized bytes of your signed transaction.
## Notes
* Send raw transaction bytes, not base64.
* Keep the QUIC connection open and reuse it for multiple transactions. This example closes after one send only to keep the sample short.
* QUIC does not support the extra gRPC submission flags or atomic bundle submission.
* If you already build transactions with `solders` or `solana-py`, serialize the signed transaction first and pass the bytes to `send_transaction()`.
See [QUIC Transaction Submission](/api/quic-transaction-submission) for the full endpoint list and protocol summary.
# Submit Single Transaction
Submit individual transactions to NextBlock using Python with proper tipping and error handling.
If you want to send raw signed transaction bytes over QUIC instead of gRPC, see [QUIC Transaction Submission](/api/examples/python/quic).
This example shows transaction construction and the request shape you send to NextBlock. Replace the placeholder generated client types with the client generated from `nextblock-proto`.
## Example
import asyncio
import base64
import random
from typing import List, Optional
from dataclasses import dataclass
from solders.pubkey import Pubkey
from solders.keypair import Keypair
from solders.system_program import TransferParams, transfer
from solders.transaction import Transaction
from solders.message import MessageV0
from solders.hash import Hash
from solana.rpc.async_api import AsyncClient
from solana.rpc.commitment import Finalized
# NextBlock tip wallets for load balancing
NEXTBLOCK_TIP_WALLETS = [
"NEXTbLoCkB51HpLBLojQfpyVAMorm3zzKg7w9NFdqid",
"nextBLoCkPMgmG8ZgJtABeScP35qLa2AMCNKntAP7Xc",
"NextbLoCkVtMGcV47JzewQdvBpLqT9TxQFozQkN98pE",
"NexTbLoCkWykbLuB1NkjXgFWkX9oAtcoagQegygXXA2",
"NeXTBLoCKs9F1y5PJS9CKrFNNLU1keHW71rfh7KgA1X",
"NexTBLockJYZ7QD7p2byrUa6df8ndV2WSd8GkbWqfbb",
"neXtBLock1LeC67jYd1QdAa32kbVeubsfPNTJC1V5At",
"nEXTBLockYgngeRmRrjDV31mGSekVPqZoMGhQEZtPVG",
]
# Get random tip wallet for load balancing
def get_random_nextblock_tip_wallet() -> Pubkey:
wallet_str = random.choice(NEXTBLOCK_TIP_WALLETS)
return Pubkey.from_string(wallet_str)
# Transaction submission parameters
@dataclass
class SubmissionOptions:
skip_preflight: bool = True
front_running_protection: bool = False
revert_on_fail: bool = False
disable_retries: bool = False
snipe_transaction: bool = False
# Build transaction with tip
async def build_transaction_with_tip(
payer: Keypair,
recent_blockhash: Hash,
tip_amount: int,
instructions: List,
) -> Transaction:
# Create tip instruction (should be first)
tip_wallet = get_random_nextblock_tip_wallet()
tip_instruction = transfer(
TransferParams(
from_pubkey=payer.pubkey(),
to_pubkey=tip_wallet,
lamports=tip_amount
)
)
# Combine all instructions
all_instructions = [tip_instruction] + instructions
# Create and sign transaction
message = MessageV0.try_compile(
payer=payer.pubkey(),
instructions=all_instructions,
address_lookup_table_accounts=[],
recent_blockhash=recent_blockhash,
)
transaction = Transaction.new_unsigned(message)
transaction.sign([payer], recent_blockhash)
return transaction
# Submit single transaction to NextBlock
async def submit_single_transaction(
# nextblock_client, # Your generated gRPC client
rpc_client: AsyncClient,
signer: Keypair,
recipient: Pubkey,
transfer_amount: int,
tip_amount: int,
options: SubmissionOptions = None,
) -> str:
if options is None:
options = SubmissionOptions()
# Get recent blockhash
response = await rpc_client.get_latest_blockhash(commitment=Finalized)
recent_blockhash = response.value.blockhash
# Create transfer instruction
transfer_instruction = transfer(
TransferParams(
from_pubkey=signer.pubkey(),
to_pubkey=recipient,
lamports=transfer_amount
)
)
# Build transaction with tip
transaction = await build_transaction_with_tip(
signer,
recent_blockhash,
tip_amount,
[transfer_instruction]
)
# Convert to base64 for submission
serialized_tx = bytes(transaction)
base64_tx = base64.b64encode(serialized_tx).decode('utf-8')
# Submit to NextBlock
""" Uncomment when you have the generated gRPC client
request = PostSubmitRequest(
transaction=TransactionMessage(content=base64_tx),
skip_pre_flight=options.skip_preflight,
snipe_transaction=options.snipe_transaction,
front_running_protection=options.front_running_protection,
disable_retries=options.disable_retries,
revert_on_fail=options.revert_on_fail,
)
response = await nextblock_client.post_submit_v2(request)
print(f"Transaction submitted successfully!")
print(f"Signature: {response.signature}")
print(f"UUID: {response.uuid}")
return response.signature
"""
# Until your generated client is wired in, return the local signature
signature = str(transaction.signatures[0])
print(f"Local transaction built successfully: {signature}")
return signature
# Submit with optimal tip calculation
async def submit_with_optimal_tip(
# nextblock_client,
rpc_client: AsyncClient,
signer: Keypair,
recipient: Pubkey,
transfer_amount: int,
priority_level: str = "normal",
) -> str:
# Calculate optimal tip based on priority level
tip_amounts = {
"conservative": 500_000, # 0.0005 SOL
"normal": 1_000_000, # 0.001 SOL
"aggressive": 2_000_000, # 0.002 SOL
"priority": 5_000_000, # 0.005 SOL
}
tip_amount = tip_amounts.get(priority_level, tip_amounts["normal"])
return await submit_single_transaction(
# nextblock_client,
rpc_client,
signer,
recipient,
transfer_amount,
tip_amount
)
# Batch multiple single transactions
async def submit_multiple_single_transactions(
# nextblock_client,
rpc_client: AsyncClient,
signer: Keypair,
transactions_data: List[tuple], # [(recipient, amount, tip), ...]
) -> List[str]:
"""Submit multiple single transactions concurrently"""
tasks = []
for recipient, transfer_amount, tip_amount in transactions_data:
task = submit_single_transaction(
# nextblock_client,
rpc_client,
signer,
recipient,
transfer_amount,
tip_amount
)
tasks.append(task)
# Execute all transactions concurrently
signatures = await asyncio.gather(*tasks, return_exceptions=True)
# Handle results
successful_signatures = []
for i, result in enumerate(signatures):
if isinstance(result, Exception):
print(f"Transaction {i+1} failed: {result}")
else:
successful_signatures.append(result)
print(f"Transaction {i+1} successful: {result}")
return successful_signatures
## Advanced Features
import time
from typing import Dict, Any
import backoff
# Transaction with retry logic
@backoff.on_exception(
backoff.expo,
Exception,
max_tries=3,
base=2,
max_value=30
)
async def submit_transaction_with_retry(
# nextblock_client,
rpc_client: AsyncClient,
signer: Keypair,
recipient: Pubkey,
transfer_amount: int,
tip_amount: int,
) -> str:
return await submit_single_transaction(
# nextblock_client,
rpc_client,
signer,
recipient,
transfer_amount,
tip_amount
)
# Transaction builder with custom instructions
class TransactionBuilder:
def __init__(self, payer: Keypair):
self.payer = payer
self.instructions = []
self.tip_amount = 1_000_000 # Default tip
def add_instruction(self, instruction):
"""Add custom instruction to transaction"""
self.instructions.append(instruction)
return self
def set_tip_amount(self, amount: int):
"""Set tip amount in lamports"""
self.tip_amount = amount
return self
async def build_and_submit(
self,
# nextblock_client,
rpc_client: AsyncClient,
options: SubmissionOptions = None
) -> str:
"""Build and submit the transaction"""
if not self.instructions:
raise ValueError("No instructions added to transaction")
# Get recent blockhash
response = await rpc_client.get_latest_blockhash(commitment=Finalized)
recent_blockhash = response.value.blockhash
# Build transaction with tip
transaction = await build_transaction_with_tip(
self.payer,
recent_blockhash,
self.tip_amount,
self.instructions
)
# Convert and submit
serialized_tx = bytes(transaction)
base64_tx = base64.b64encode(serialized_tx).decode('utf-8')
# Submit logic here (similar to above)
signature = str(transaction.signatures[0])
print(f"Custom transaction submitted: {signature}")
return signature
# Performance monitoring
class TransactionMetrics:
def __init__(self):
self.submissions = 0
self.successes = 0
self.failures = 0
self.total_time = 0.0
self.start_time = None
def start_submission(self):
self.submissions += 1
self.start_time = time.time()
def record_success(self):
if self.start_time:
self.total_time += time.time() - self.start_time
self.successes += 1
self.start_time = None
def record_failure(self):
if self.start_time:
self.total_time += time.time() - self.start_time
self.failures += 1
self.start_time = None
def get_stats(self) -> Dict[str, Any]:
return {
"total_submissions": self.submissions,
"successes": self.successes,
"failures": self.failures,
"success_rate": self.successes / max(self.submissions, 1) * 100,
"average_time": self.total_time / max(self.successes, 1),
}
# Monitored transaction submission
async def submit_with_monitoring(
# nextblock_client,
rpc_client: AsyncClient,
signer: Keypair,
recipient: Pubkey,
transfer_amount: int,
tip_amount: int,
metrics: TransactionMetrics,
) -> Optional[str]:
metrics.start_submission()
try:
signature = await submit_single_transaction(
# nextblock_client,
rpc_client,
signer,
recipient,
transfer_amount,
tip_amount
)
metrics.record_success()
return signature
except Exception as e:
metrics.record_failure()
print(f"Transaction failed: {e}")
return None
## Usage Examples
async def main():
# Initialize clients
rpc_client = AsyncClient("https://api.mainnet-beta.solana.com")
signer = Keypair() # Use your actual keypair
recipient = Pubkey.from_string("<recipient-public-key>")
# Connect to NextBlock (see connection.md)
# config = NextBlockConfig.from_env()
# async with NextBlockConnectionManager(config) as manager:
# nextblock_client = manager.client
try:
# Example 1: Basic transaction submission
signature1 = await submit_single_transaction(
# nextblock_client,
rpc_client,
signer,
recipient,
10_000, # Transfer 10,000 lamports
1_000_000, # Tip 1,000,000 lamports (0.001 SOL)
)
print(f"Basic transaction: {signature1}")
# Example 2: Transaction with optimal tip
signature2 = await submit_with_optimal_tip(
# nextblock_client,
rpc_client,
signer,
recipient,
20_000, # Transfer 20,000 lamports
"aggressive" # Use aggressive tip strategy
)
print(f"Optimally tipped transaction: {signature2}")
# Example 3: Custom transaction builder
builder = TransactionBuilder(signer)
signature3 = await (builder
.add_instruction(transfer(TransferParams(
from_pubkey=signer.pubkey(),
to_pubkey=recipient,
lamports=30_000
)))
.set_tip_amount(1_500_000)
.build_and_submit(
# nextblock_client,
rpc_client
))
print(f"Custom built transaction: {signature3}")
# Example 4: Multiple transactions
transactions_data = [
(recipient, 5_000, 500_000),
(recipient, 7_500, 750_000),
(recipient, 12_500, 1_250_000),
]
signatures = await submit_multiple_single_transactions(
# nextblock_client,
rpc_client,
signer,
transactions_data
)
print(f"Multiple transactions: {signatures}")
# Example 5: Transaction with monitoring
metrics = TransactionMetrics()
for i in range(5):
await submit_with_monitoring(
# nextblock_client,
rpc_client,
signer,
recipient,
1_000 * (i + 1), # Varying amounts
500_000, # Fixed tip
metrics
)
print(f"Performance metrics: {metrics.get_stats()}")
finally:
await rpc_client.close()
# Run the examples
if __name__ == "__main__":
asyncio.run(main())
## Best Practices
1. **Always include tips**: NextBlock prioritizes transactions with appropriate tips
2. **Use random tip wallets**: Distribute load across multiple tip addresses
3. **Monitor tip floors**: Adjust tips based on current network conditions
4. **Handle errors gracefully**: Implement retry logic with exponential backoff
5. **Validate inputs**: Always validate public keys and amounts before submission
6. **Use async/await**: Leverage Python's asyncio for better performance
7. **Monitor performance**: Track success rates and response times
8. **Choose appropriate RPC endpoints**: Use reliable RPC providers for blockhash retrieval
# Submit Batched Transactions
Submit 2-4 transactions as an atomic bundle to NextBlock using Python. Batched transactions are processed as Jito bundles - either all succeed or none do.
This example shows bundle construction and the request shape you send to NextBlock. Replace the placeholder generated client types with the client generated from `nextblock-proto`.
## Example
import asyncio
import base64
import random
from typing import List, Optional, Tuple
from dataclasses import dataclass
from solders.pubkey import Pubkey
from solders.keypair import Keypair
from solders.system_program import TransferParams, transfer
from solders.transaction import Transaction
from solders.message import MessageV0
from solders.hash import Hash
from solana.rpc.async_api import AsyncClient
from solana.rpc.commitment import Finalized
# NextBlock tip wallets
NEXTBLOCK_TIP_WALLETS = [
"NEXTbLoCkB51HpLBLojQfpyVAMorm3zzKg7w9NFdqid",
"nextBLoCkPMgmG8ZgJtABeScP35qLa2AMCNKntAP7Xc",
"NextbLoCkVtMGcV47JzewQdvBpLqT9TxQFozQkN98pE",
"NexTbLoCkWykbLuB1NkjXgFWkX9oAtcoagQegygXXA2",
"NeXTBLoCKs9F1y5PJS9CKrFNNLU1keHW71rfh7KgA1X",
"NexTBLockJYZ7QD7p2byrUa6df8ndV2WSd8GkbWqfbb",
"neXtBLock1LeC67jYd1QdAa32kbVeubsfPNTJC1V5At",
"nEXTBLockYgngeRmRrjDV31mGSekVPqZoMGhQEZtPVG",
]
# Transaction bundle builder
class TransactionBundle:
def __init__(self):
self.transactions: List[Transaction] = []
self.max_size = 4
self.min_size = 2
def add_transaction(self, transaction: Transaction) -> 'TransactionBundle':
"""Add transaction to bundle"""
if len(self.transactions) >= self.max_size:
raise ValueError(f"Bundle cannot contain more than {self.max_size} transactions")
self.transactions.append(transaction)
return self
def validate(self) -> bool:
"""Validate bundle before submission"""
if len(self.transactions) < self.min_size:
raise ValueError(f"Bundle must contain at least {self.min_size} transactions")
if len(self.transactions) > self.max_size:
raise ValueError(f"Bundle cannot contain more than {self.max_size} transactions")
# Check for duplicate signatures
signatures = set()
for tx in self.transactions:
sig = str(tx.signatures[0])
if sig in signatures:
raise ValueError("Duplicate transaction signatures in bundle")
signatures.add(sig)
return True
def to_base64_transactions(self) -> List[str]:
"""Convert all transactions to base64 strings"""
self.validate()
base64_transactions = []
for tx in self.transactions:
serialized_tx = bytes(tx)
base64_tx = base64.b64encode(serialized_tx).decode('utf-8')
base64_transactions.append(base64_tx)
return base64_transactions
def get_signatures(self) -> List[str]:
"""Get all transaction signatures"""
return [str(tx.signatures[0]) for tx in self.transactions]
# Build single transaction with tip
async def build_transaction_with_tip(
payer: Keypair,
recent_blockhash: Hash,
tip_amount: int,
instructions: List,
) -> Transaction:
# Random tip wallet for load balancing
tip_wallet = Pubkey.from_string(random.choice(NEXTBLOCK_TIP_WALLETS))
# Create tip instruction (should be first)
tip_instruction = transfer(
TransferParams(
from_pubkey=payer.pubkey(),
to_pubkey=tip_wallet,
lamports=tip_amount
)
)
# Combine all instructions
all_instructions = [tip_instruction] + instructions
# Create and sign transaction
message = MessageV0.try_compile(
payer=payer.pubkey(),
instructions=all_instructions,
address_lookup_table_accounts=[],
recent_blockhash=recent_blockhash,
)
transaction = Transaction.new_unsigned(message)
transaction.sign([payer], recent_blockhash)
return transaction
# Submit batched transactions
async def submit_batched_transactions(
# nextblock_client, # Your generated gRPC client
rpc_client: AsyncClient,
signer: Keypair,
transaction_specs: List[Tuple[List, int]], # [(instructions, tip_amount), ...]
) -> str:
"""Submit multiple transactions as an atomic bundle"""
# Get recent blockhash (same for all transactions in bundle)
response = await rpc_client.get_latest_blockhash(commitment=Finalized)
recent_blockhash = response.value.blockhash
# Build transaction bundle
bundle = TransactionBundle()
for instructions, tip_amount in transaction_specs:
transaction = await build_transaction_with_tip(
signer,
recent_blockhash,
tip_amount,
instructions
)
bundle.add_transaction(transaction)
# Get base64 transactions for submission
base64_transactions = bundle.to_base64_transactions()
signatures = bundle.get_signatures()
# Log transaction signatures
for i, sig in enumerate(signatures):
print(f"Transaction {i+1} signature: {sig}")
# Submit bundle to NextBlock
""" Uncomment when you have the generated gRPC client
entries = [
PostSubmitRequestEntry(
transaction=TransactionMessage(content=base64_tx)
)
for base64_tx in base64_transactions
]
request = PostSubmitBatchRequest(entries=entries)
response = await nextblock_client.post_submit_batch_v2(request)
print(f"Batch submitted successfully!")
print(f"Bundle signature: {response.signature}")
return response.signature
"""
# Until your generated client is wired in, return a placeholder bundle signature
bundle_signature = ""
print(f"Local bundle prepared with {len(bundle.transactions)} transactions.")
return bundle_signature
# Build common transaction patterns
async def build_arbitrage_bundle(
# nextblock_client,
rpc_client: AsyncClient,
signer: Keypair,
dex_a_address: Pubkey,
dex_b_address: Pubkey,
trade_amount: int,
) -> str:
"""Build arbitrage bundle for cross-DEX trading"""
# Transaction 1: Buy on DEX A
buy_instructions = [
# Add your DEX-specific buy instructions here
transfer(TransferParams(
from_pubkey=signer.pubkey(),
to_pubkey=dex_a_address,
lamports=trade_amount
))
]
# Transaction 2: Sell on DEX B
sell_instructions = [
# Add your DEX-specific sell instructions here
transfer(TransferParams(
from_pubkey=signer.pubkey(),
to_pubkey=dex_b_address,
lamports=trade_amount
))
]
transaction_specs = [
(buy_instructions, 2_000_000), # Higher tip for arbitrage
(sell_instructions, 2_000_000), # Higher tip for arbitrage
]
return await submit_batched_transactions(
# nextblock_client,
rpc_client,
signer,
transaction_specs
)
# Build complex DeFi operation bundle
async def build_defi_operation_bundle(
# nextblock_client,
rpc_client: AsyncClient,
signer: Keypair,
) -> str:
"""Build complex DeFi operation bundle"""
# Transaction 1: Setup - Create token accounts
setup_instructions = [
# Add token account creation instructions
transfer(TransferParams(
from_pubkey=signer.pubkey(),
to_pubkey=Pubkey.from_string("<token-program-address>"),
lamports=1_000_000 # Rent exemption
))
]
# Transaction 2: Main operation - Execute swap
swap_instructions = [
# Add swap instructions (Jupiter, Raydium, etc.)
transfer(TransferParams(
from_pubkey=signer.pubkey(),
to_pubkey=Pubkey.from_string("<swap-program-address>"),
lamports=0 # No SOL transfer for swap
))
]
# Transaction 3: Cleanup - Stake or provide liquidity
stake_instructions = [
# Add staking/liquidity instructions
transfer(TransferParams(
from_pubkey=signer.pubkey(),
to_pubkey=Pubkey.from_string("<stake-pool-address>"),
lamports=0
))
]
transaction_specs = [
(setup_instructions, 500_000), # Setup tip
(swap_instructions, 1_500_000), # Main operation tip
(stake_instructions, 750_000), # Cleanup tip
]
return await submit_batched_transactions(
# nextblock_client,
rpc_client,
signer,
transaction_specs
)
## Advanced Bundle Management
# Bundle with conditional execution
class ConditionalBundle(TransactionBundle):
def __init__(self):
super().__init__()
self.conditions: List[callable] = []
def add_conditional_transaction(
self,
transaction: Transaction,
condition: callable
) -> 'ConditionalBundle':
"""Add transaction that only executes if condition is met"""
self.add_transaction(transaction)
self.conditions.append(condition)
return self
async def evaluate_conditions(self, rpc_client: AsyncClient) -> List[bool]:
"""Evaluate all conditions before submission"""
results = []
for condition in self.conditions:
try:
result = await condition(rpc_client) if asyncio.iscoroutinefunction(condition) else condition()
results.append(bool(result))
except Exception as e:
print(f"Condition evaluation failed: {e}")
results.append(False)
return results
def filter_by_conditions(self, condition_results: List[bool]) -> 'TransactionBundle':
"""Create new bundle with only transactions that meet conditions"""
filtered_bundle = TransactionBundle()
for i, (transaction, condition_met) in enumerate(zip(self.transactions, condition_results)):
if condition_met:
filtered_bundle.add_transaction(transaction)
else:
print(f"Transaction {i+1} filtered out due to condition")
return filtered_bundle
# Bundle performance optimizer
class BundleOptimizer:
def __init__(self):
self.tip_multipliers = {
"setup": 0.5, # Lower priority
"main": 1.5, # Higher priority
"cleanup": 0.75, # Medium priority
"arbitrage": 2.0, # Highest priority
}
def optimize_tips(
self,
base_tip: int,
transaction_types: List[str]
) -> List[int]:
"""Optimize tip amounts based on transaction types"""
optimized_tips = []
for tx_type in transaction_types:
multiplier = self.tip_multipliers.get(tx_type, 1.0)
optimized_tip = int(base_tip * multiplier)
optimized_tips.append(optimized_tip)
return optimized_tips
def reorder_transactions(
self,
transactions: List[Transaction],
transaction_types: List[str]
) -> Tuple[List[Transaction], List[str]]:
"""Reorder transactions for optimal execution"""
# Priority order: setup -> main -> cleanup
priority_order = {"setup": 1, "main": 2, "cleanup": 3, "arbitrage": 0}
# Create pairs and sort by priority
tx_pairs = list(zip(transactions, transaction_types))
tx_pairs.sort(key=lambda x: priority_order.get(x[1], 2))
# Unzip the sorted pairs
sorted_transactions, sorted_types = zip(*tx_pairs)
return list(sorted_transactions), list(sorted_types)
# Bundle status monitoring
@dataclass
class BundleStatus:
bundle_id: str
transaction_count: int
submitted_at: float
signatures: List[str]
status: str = "pending" # pending, confirmed, failed
def is_complete(self) -> bool:
return self.status in ["confirmed", "failed"]
class BundleTracker:
def __init__(self):
self.bundles: Dict[str, BundleStatus] = {}
def track_bundle(self, bundle_status: BundleStatus):
"""Start tracking a bundle"""
self.bundles[bundle_status.bundle_id] = bundle_status
async def check_bundle_status(
self,
bundle_id: str,
rpc_client: AsyncClient
) -> Optional[BundleStatus]:
"""Check the status of a tracked bundle"""
if bundle_id not in self.bundles:
return None
bundle_status = self.bundles[bundle_id]
# Check if all transactions are confirmed
confirmed_count = 0
for signature in bundle_status.signatures:
try:
# Check transaction status
# response = await rpc_client.get_signature_status(signature)
# if response.value and response.value.confirmation_status:
# confirmed_count += 1
confirmed_count += 1 # Mock confirmation
except Exception as e:
print(f"Failed to check signature {signature}: {e}")
# Update bundle status
if confirmed_count == bundle_status.transaction_count:
bundle_status.status = "confirmed"
elif time.time() - bundle_status.submitted_at > 60: # Timeout after 60 seconds
bundle_status.status = "failed"
return bundle_status
## Usage Examples
async def main():
# Initialize clients
rpc_client = AsyncClient("https://api.mainnet-beta.solana.com")
signer = Keypair() # Use your actual keypair
# Connect to NextBlock (see connection.md)
# config = NextBlockConfig.from_env()
# async with NextBlockConnectionManager(config) as manager:
# nextblock_client = manager.client
try:
# Example 1: Basic batch submission
transaction_specs = [
# Setup transaction
([transfer(TransferParams(
from_pubkey=signer.pubkey(),
to_pubkey=Pubkey.from_string("<recipient1>"),
lamports=100_000
))], 500_000), # 0.0005 SOL tip
# Main transaction
([transfer(TransferParams(
from_pubkey=signer.pubkey(),
to_pubkey=Pubkey.from_string("<recipient2>"),
lamports=200_000
))], 1_000_000), # 0.001 SOL tip
# Cleanup transaction
([transfer(TransferParams(
from_pubkey=signer.pubkey(),
to_pubkey=Pubkey.from_string("<recipient3>"),
lamports=50_000
))], 500_000), # 0.0005 SOL tip
]
bundle_signature = await submit_batched_transactions(
# nextblock_client,
rpc_client,
signer,
transaction_specs
)
print(f"Basic batch: {bundle_signature}")
# Example 2: Arbitrage bundle
arb_signature = await build_arbitrage_bundle(
# nextblock_client,
rpc_client,
signer,
Pubkey.from_string("<dex-a-address>"),
Pubkey.from_string("<dex-b-address>"),
1_000_000 # 0.001 SOL trade
)
print(f"Arbitrage bundle: {arb_signature}")
# Example 3: Optimized DeFi bundle
defi_signature = await build_defi_operation_bundle(
# nextblock_client,
rpc_client,
signer
)
print(f"DeFi bundle: {defi_signature}")
# Example 4: Bundle with optimization
optimizer = BundleOptimizer()
base_tip = 1_000_000
transaction_types = ["setup", "main", "cleanup"]
optimized_tips = optimizer.optimize_tips(base_tip, transaction_types)
print(f"Optimized tips: {optimized_tips}")
finally:
await rpc_client.close()
if __name__ == "__main__":
asyncio.run(main())
## Best Practices
1. **Bundle size limits**: Keep bundles between 2-4 transactions for optimal success rates
2. **Transaction ordering**: Setup → Main operations → Cleanup
3. **Progressive tipping**: Use higher tips for more critical transactions
4. **Error handling**: Validate bundles before submission
5. **Performance monitoring**: Track bundle success rates and timing
6. **Conditional execution**: Filter transactions based on current conditions
7. **Tip optimization**: Adjust tips based on transaction importance and network conditions
# Tip Floor Stream
Stream real-time tip floor data from NextBlock to optimize transaction tips dynamically using Python.
import asyncio
import json
import time
from typing import Dict, Any, Optional, List
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
import logging
# Tip floor data structure
@dataclass
class TipFloorData:
time: str
landed_tips_25th_percentile: float
landed_tips_50th_percentile: float
landed_tips_75th_percentile: float
landed_tips_95th_percentile: float
landed_tips_99th_percentile: float
ema_landed_tips_50th_percentile: float
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'TipFloorData':
return cls(**data)
def to_lamports(self, percentile: str) -> int:
"""Convert SOL amounts to lamports"""
sol_amount = getattr(self, f"landed_tips_{percentile}_percentile")
return int(sol_amount * 1_000_000_000)
# Tip strategy management
@dataclass
class TipStrategy:
conservative_tip: int = 500_000 # 25th percentile
normal_tip: int = 1_000_000 # 50th percentile
aggressive_tip: int = 2_000_000 # 75th percentile
priority_tip: int = 5_000_000 # 95th percentile
last_updated: datetime = None
def update_from_tip_floor(self, tip_floor: TipFloorData):
"""Update strategy based on tip floor data"""
self.conservative_tip = tip_floor.to_lamports("25th")
self.normal_tip = tip_floor.to_lamports("50th")
self.aggressive_tip = tip_floor.to_lamports("75th")
self.priority_tip = tip_floor.to_lamports("95th")
self.last_updated = datetime.now(timezone.utc)
def get_tip_for_priority(self, priority: str) -> int:
"""Get tip amount for given priority level"""
return {
"conservative": self.conservative_tip,
"normal": self.normal_tip,
"aggressive": self.aggressive_tip,
"priority": self.priority_tip,
}.get(priority, self.normal_tip)
# Global tip strategy instance
global_tip_strategy = TipStrategy()
# Stream tip floor data
async def stream_tip_floor(
# nextblock_client, # Your generated gRPC client
update_frequency: str = "1m",
callback: Optional[callable] = None,
) -> None:
"""Stream tip floor updates from NextBlock"""
print(f"Starting tip floor stream with frequency: {update_frequency}")
""" Uncomment when you have the generated gRPC client
try:
request = TipFloorStreamRequest(update_frequency=update_frequency)
stream = nextblock_client.stream_tip_floor(request)
print("Streaming tip floor data:")
async for tip_floor_response in stream:
try:
# Convert protobuf response to TipFloorData
tip_floor = TipFloorData(
time=tip_floor_response.time,
landed_tips_25th_percentile=tip_floor_response.landed_tips_25th_percentile,
landed_tips_50th_percentile=tip_floor_response.landed_tips_50th_percentile,
landed_tips_75th_percentile=tip_floor_response.landed_tips_75th_percentile,
landed_tips_95th_percentile=tip_floor_response.landed_tips_95th_percentile,
landed_tips_99th_percentile=tip_floor_response.landed_tips_99th_percentile,
ema_landed_tips_50th_percentile=tip_floor_response.ema_landed_tips_50th_percentile,
)
print(f"Received tip floor update:")
print(f" Time: {tip_floor.time}")
print(f" 25th percentile: {tip_floor.landed_tips_25th_percentile:.6f} SOL")
print(f" 50th percentile: {tip_floor.landed_tips_50th_percentile:.6f} SOL")
print(f" 75th percentile: {tip_floor.landed_tips_75th_percentile:.6f} SOL")
print(f" 95th percentile: {tip_floor.landed_tips_95th_percentile:.6f} SOL")
print(f" EMA 50th percentile: {tip_floor.ema_landed_tips_50th_percentile:.6f} SOL")
print(" ---")
# Update global tip strategy
await process_tip_floor_update(tip_floor)
# Call custom callback if provided
if callback:
await callback(tip_floor)
except Exception as e:
logging.error(f"Error processing tip floor update: {e}")
except Exception as e:
logging.error(f"Tip floor stream error: {e}")
# Implement reconnection logic here
"""
# Mock streaming for demonstration
print("Mock tip floor streaming started...")
while True:
await asyncio.sleep(60) # Update every minute
# Generate mock tip floor data
mock_tip_floor = TipFloorData(
time=datetime.now(timezone.utc).isoformat(),
landed_tips_25th_percentile=0.0011,
landed_tips_50th_percentile=0.005000001,
landed_tips_75th_percentile=0.01555,
landed_tips_95th_percentile=0.09339195639999975,
landed_tips_99th_percentile=0.4846427910400001,
ema_landed_tips_50th_percentile=0.005989477267191758,
)
print(f"Mock tip floor update: {asdict(mock_tip_floor)}")
await process_tip_floor_update(mock_tip_floor)
if callback:
await callback(mock_tip_floor)
# Process tip floor updates
async def process_tip_floor_update(tip_floor: TipFloorData) -> None:
"""Process incoming tip floor data"""
# Update global strategy
global_tip_strategy.update_from_tip_floor(tip_floor)
# Log the update
logging.info(f"Updated tip strategy at {tip_floor.time}")
logging.info(f" Conservative: {global_tip_strategy.conservative_tip} lamports")
logging.info(f" Normal: {global_tip_strategy.normal_tip} lamports")
logging.info(f" Aggressive: {global_tip_strategy.aggressive_tip} lamports")
logging.info(f" Priority: {global_tip_strategy.priority_tip} lamports")
# Store historical data
await store_tip_floor_data(tip_floor)
# Trigger any pending transactions
await trigger_pending_transactions()
# Historical data management
class TipFloorHistory:
def __init__(self, max_size: int = 1000):
self.data: List[TipFloorData] = []
self.max_size = max_size
def add(self, tip_floor: TipFloorData):
"""Add tip floor data to history"""
if len(self.data) >= self.max_size:
self.data.pop(0) # Remove oldest
self.data.append(tip_floor)
def get_trend(self, percentile: str = "50th", window: int = 10) -> float:
"""Calculate tip trend over time window"""
if len(self.data) < 2:
return 0.0
recent_data = self.data[-window:] if len(self.data) >= window else self.data
if len(recent_data) < 2:
return 0.0
start_value = getattr(recent_data[0], f"landed_tips_{percentile}_percentile")
end_value = getattr(recent_data[-1], f"landed_tips_{percentile}_percentile")
return end_value - start_value
def get_average(self, percentile: str = "50th", window: int = 10) -> float:
"""Get average tip over time window"""
if not self.data:
return 0.0
recent_data = self.data[-window:] if len(self.data) >= window else self.data
values = [getattr(d, f"landed_tips_{percentile}_percentile") for d in recent_data]
return sum(values) / len(values)
# Global history tracker
tip_floor_history = TipFloorHistory()
# Smart tip calculation with trend analysis
async def get_smart_tip(
base_priority: str = "normal",
consider_trend: bool = True,
) -> int:
"""Calculate smart tip amount based on current data and trends"""
base_tip = global_tip_strategy.get_tip_for_priority(base_priority)
if not consider_trend or len(tip_floor_history.data) < 2:
return base_tip
# Analyze trend
trend = tip_floor_history.get_trend("50th", window=5)
# Adjust tip based on trend
if trend > 0.001: # Tips increasing
adjustment_factor = 1.2
print(f"Tips trending up (+{trend:.6f}), increasing tip by 20%")
elif trend < -0.001: # Tips decreasing
adjustment_factor = 0.9
print(f"Tips trending down ({trend:.6f}), decreasing tip by 10%")
else:
adjustment_factor = 1.0
print(f"Tips stable ({trend:.6f}), no adjustment")
smart_tip = int(base_tip * adjustment_factor)
return max(smart_tip, 100_000) # Minimum tip of 0.0001 SOL
# Store tip floor data
async def store_tip_floor_data(tip_floor: TipFloorData) -> None:
"""Store tip floor data for analysis"""
# Add to history
tip_floor_history.add(tip_floor)
# Optionally save to file
filename = f"tip_data_{datetime.now().strftime('%Y%m%d')}.jsonl"
with open(filename, "a") as f:
json.dump(asdict(tip_floor), f)
f.write("\n")
# Trigger pending transactions
async def trigger_pending_transactions() -> None:
"""Check and trigger any pending transactions with updated tips"""
print("Checking for pending transactions to trigger...")
# Implementation would check your pending transaction queue
# and submit them with updated tip amounts
## Usage Example
async def tip_floor_example():
# Connect to NextBlock (see connection.md)
# config = NextBlockConfig.from_env()
# async with NextBlockConnectionManager(config) as manager:
# nextblock_client = manager.client
# Custom callback for tip floor updates
async def on_tip_floor_update(tip_floor: TipFloorData):
print(f"Custom handler: Received update at {tip_floor.time}")
# Example: Trigger high-priority transactions when tips are low
if tip_floor.landed_tips_50th_percentile < 0.002: # Less than 0.002 SOL
print("Tips are low - good time for high-priority transactions!")
# await submit_priority_transactions()
# Start streaming in background
stream_task = asyncio.create_task(
stream_tip_floor(
# nextblock_client,
update_frequency="1m",
callback=on_tip_floor_update
)
)
# Example usage of dynamic tips
await asyncio.sleep(5) # Wait for initial data
# Get current optimal tips
conservative_tip = global_tip_strategy.get_tip_for_priority("conservative")
normal_tip = global_tip_strategy.get_tip_for_priority("normal")
aggressive_tip = global_tip_strategy.get_tip_for_priority("aggressive")
print(f"Current optimal tips:")
print(f" Conservative: {conservative_tip} lamports")
print(f" Normal: {normal_tip} lamports")
print(f" Aggressive: {aggressive_tip} lamports")
# Get smart tip with trend analysis
smart_tip = await get_smart_tip("normal", consider_trend=True)
print(f" Smart tip: {smart_tip} lamports")
# Keep streaming
try:
await stream_task
except KeyboardInterrupt:
stream_task.cancel()
print("Tip floor streaming stopped")
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
asyncio.run(tip_floor_example())
# Keepalive
Maintain persistent gRPC connections to NextBlock for optimal performance using Python's asyncio.
import asyncio
import time
import logging
from typing import Optional
from dataclasses import dataclass, field
from datetime import datetime, timezone
# Connection health tracking
@dataclass
class ConnectionHealth:
is_healthy: bool = True
last_successful_ping: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
consecutive_failures: int = 0
total_pings_sent: int = 0
total_pings_successful: int = 0
average_ping_time: float = 0.0
def success_rate(self) -> float:
"""Calculate ping success rate percentage"""
if self.total_pings_sent == 0:
return 100.0
return (self.total_pings_successful / self.total_pings_sent) * 100.0
def update_success(self, ping_time: float):
"""Update health after successful ping"""
self.is_healthy = True
self.last_successful_ping = datetime.now(timezone.utc)
self.consecutive_failures = 0
self.total_pings_sent += 1
self.total_pings_successful += 1
# Update average ping time
if self.total_pings_successful == 1:
self.average_ping_time = ping_time
else:
self.average_ping_time = (
(self.average_ping_time * (self.total_pings_successful - 1) + ping_time)
/ self.total_pings_successful
)
def update_failure(self, max_failures: int = 3):
"""Update health after failed ping"""
self.consecutive_failures += 1
self.total_pings_sent += 1
if self.consecutive_failures >= max_failures:
self.is_healthy = False
# Keepalive configuration
@dataclass
class KeepaliveConfig:
ping_interval: float = 60.0 # seconds
max_consecutive_failures: int = 3
reconnect_delay: float = 5.0 # seconds
health_check_enabled: bool = True
timeout: float = 15.0 # seconds
# Basic keepalive implementation
async def start_keepalive_task(
# nextblock_client, # Your generated gRPC client
config: KeepaliveConfig = None,
) -> None:
"""Start basic keepalive task"""
if config is None:
config = KeepaliveConfig()
print(f"Starting keepalive task with {config.ping_interval}s interval")
while True:
try:
await asyncio.sleep(config.ping_interval)
# Send ping request
start_time = time.time()
""" Uncomment when you have the generated gRPC client
try:
await asyncio.wait_for(
nextblock_client.ping(Empty()),
timeout=config.timeout
)
ping_time = time.time() - start_time
print(f"Keepalive ping successful ({ping_time*1000:.1f}ms) at {datetime.now().strftime('%H:%M:%S')}")
except asyncio.TimeoutError:
print(f"Keepalive ping timeout after {config.timeout}s")
except Exception as e:
print(f"Keepalive ping failed: {e}")
# Optionally implement reconnection logic
break
"""
# Mock ping for demonstration
await asyncio.sleep(0.1) # Simulate network delay
ping_time = time.time() - start_time
print(f"Mock keepalive ping successful ({ping_time*1000:.1f}ms) at {datetime.now().strftime('%H:%M:%S')}")
except Exception as e:
logging.error(f"Keepalive task error: {e}")
break
# Advanced keepalive manager
class KeepaliveManager:
def __init__(
self,
# nextblock_client, # Your generated gRPC client
config: KeepaliveConfig = None,
):
# self.client = nextblock_client
self.config = config or KeepaliveConfig()
self.health = ConnectionHealth()
self.keepalive_task: Optional[asyncio.Task] = None
self.is_running = False
async def start(self) -> None:
"""Start the keepalive manager"""
if self.is_running:
return
self.is_running = True
self.keepalive_task = asyncio.create_task(self._keepalive_loop())
print("Keepalive manager started")
async def stop(self) -> None:
"""Stop the keepalive manager"""
self.is_running = False
if self.keepalive_task and not self.keepalive_task.done():
self.keepalive_task.cancel()
try:
await self.keepalive_task
except asyncio.CancelledError:
pass
print("Keepalive manager stopped")
async def _keepalive_loop(self) -> None:
"""Main keepalive loop"""
while self.is_running:
try:
await asyncio.sleep(self.config.ping_interval)
await self._send_ping()
except asyncio.CancelledError:
break
except Exception as e:
logging.error(f"Keepalive loop error: {e}")
# Handle connection recovery
if not self.health.is_healthy:
await self._handle_connection_recovery()
async def _send_ping(self) -> None:
"""Send a single ping and update health"""
start_time = time.time()
try:
""" Uncomment when you have the generated gRPC client
await asyncio.wait_for(
self.client.ping(Empty()),
timeout=self.config.timeout
)
"""
# Mock ping delay
await asyncio.sleep(0.05 + (time.time() % 0.1)) # Variable delay 50-150ms
ping_time = time.time() - start_time
self.health.update_success(ping_time)
print(f"Keepalive ping successful ({ping_time*1000:.1f}ms) - "
f"Health: {self.health.success_rate():.1f}%")
except asyncio.TimeoutError:
self.health.update_failure(self.config.max_consecutive_failures)
logging.warning(f"Keepalive ping timeout after {self.config.timeout}s")
except Exception as e:
self.health.update_failure(self.config.max_consecutive_failures)
logging.error(f"Keepalive ping failed: {e}")
async def _handle_connection_recovery(self) -> None:
"""Handle connection recovery when unhealthy"""
logging.warning("Connection unhealthy, attempting recovery...")
# Wait before attempting recovery
await asyncio.sleep(self.config.reconnect_delay)
# Try to recover connection
try:
# Implement connection recovery logic here
# await self._reconnect()
print("Connection recovery attempted")
except Exception as e:
logging.error(f"Connection recovery failed: {e}")
def get_health(self) -> ConnectionHealth:
"""Get current connection health"""
return self.health
def is_healthy(self) -> bool:
"""Check if connection is healthy"""
return self.health.is_healthy
# Connection manager with integrated keepalive
class ConnectionManagerWithKeepalive:
def __init__(self, nextblock_config, keepalive_config: KeepaliveConfig = None):
self.nextblock_config = nextblock_config
self.keepalive_config = keepalive_config or KeepaliveConfig()
# self.client = None
self.keepalive_manager: Optional[KeepaliveManager] = None
self.is_connected = False
async def connect(self) -> bool:
"""Establish connection and start keepalive"""
try:
# Create connection (see connection.md)
# channel, client = await create_nextblock_client(self.nextblock_config)
# self.client = client
# Test connection
# await self.client.ping(Empty())
self.is_connected = True
print("Successfully connected to NextBlock")
# Start keepalive
self.keepalive_manager = KeepaliveManager(
# self.client,
self.keepalive_config
)
await self.keepalive_manager.start()
return True
except Exception as e:
logging.error(f"Failed to connect: {e}")
self.is_connected = False
return False
async def disconnect(self) -> None:
"""Disconnect and stop keepalive"""
if self.keepalive_manager:
await self.keepalive_manager.stop()
self.is_connected = False
print("Disconnected from NextBlock")
def get_connection_health(self) -> Optional[ConnectionHealth]:
"""Get current connection health"""
if self.keepalive_manager:
return self.keepalive_manager.get_health()
return None
async def __aenter__(self):
await self.connect()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.disconnect()
# Health monitoring and alerting
class HealthMonitor:
def __init__(self, keepalive_manager: KeepaliveManager):
self.keepalive_manager = keepalive_manager
self.alert_thresholds = {
"success_rate": 90.0, # Alert if success rate < 90%
"avg_ping_time": 1.0, # Alert if avg ping time > 1s
"consecutive_failures": 2, # Alert after 2 consecutive failures
}
async def start_monitoring(self, check_interval: float = 30.0) -> None:
"""Start health monitoring with periodic checks"""
while True:
await asyncio.sleep(check_interval)
await self._check_health()
async def _check_health(self) -> None:
"""Check connection health and trigger alerts if needed"""
health = self.keepalive_manager.get_health()
# Check success rate
if health.success_rate() < self.alert_thresholds["success_rate"]:
await self._trigger_alert(
"Low success rate",
f"Success rate: {health.success_rate():.1f}%"
)
# Check average ping time
if health.average_ping_time > self.alert_thresholds["avg_ping_time"]:
await self._trigger_alert(
"High ping time",
f"Average ping time: {health.average_ping_time*1000:.1f}ms"
)
# Check consecutive failures
if health.consecutive_failures >= self.alert_thresholds["consecutive_failures"]:
await self._trigger_alert(
"Connection issues",
f"Consecutive failures: {health.consecutive_failures}"
)
async def _trigger_alert(self, alert_type: str, details: str) -> None:
"""Trigger health alert"""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"HEALTH ALERT [{timestamp}] {alert_type}: {details}")
# Implement additional alerting logic here
# - Send email notifications
# - Post to Slack/Discord
# - Write to monitoring system
## Usage Examples
async def basic_keepalive_example():
"""Basic keepalive usage"""
# Connect to NextBlock (see connection.md)
# config = NextBlockConfig.from_env()
# channel, client = await create_nextblock_client(config)
# Start keepalive task
keepalive_task = asyncio.create_task(
start_keepalive_task(
# client,
KeepaliveConfig(ping_interval=60.0)
)
)
# Your main application logic here
print("Application running with keepalive...")
try:
# Simulate application work
await asyncio.sleep(300) # Run for 5 minutes
except KeyboardInterrupt:
print("Stopping application...")
finally:
keepalive_task.cancel()
try:
await keepalive_task
except asyncio.CancelledError:
pass
async def advanced_keepalive_example():
"""Advanced keepalive with health monitoring"""
# from connection import NextBlockConfig
# Configuration
# nextblock_config = NextBlockConfig.from_env()
keepalive_config = KeepaliveConfig(
ping_interval=30.0, # Ping every 30 seconds
max_consecutive_failures=3,
reconnect_delay=10.0,
timeout=15.0,
)
# Use connection manager with integrated keepalive
async with ConnectionManagerWithKeepalive(
# nextblock_config,
None, # Placeholder
keepalive_config
) as manager:
if manager.is_connected:
print("Connected with keepalive enabled")
# Start health monitoring
health_monitor = HealthMonitor(manager.keepalive_manager)
monitor_task = asyncio.create_task(
health_monitor.start_monitoring(check_interval=60.0)
)
try:
# Your application logic here
for i in range(10):
await asyncio.sleep(30)
health = manager.get_connection_health()
if health:
print(f"Connection health check {i+1}:")
print(f" Healthy: {health.is_healthy}")
print(f" Success rate: {health.success_rate():.1f}%")
print(f" Avg ping time: {health.average_ping_time*1000:.1f}ms")
print(f" Total pings: {health.total_pings_sent}")
except KeyboardInterrupt:
print("Application interrupted")
finally:
monitor_task.cancel()
try:
await monitor_task
except asyncio.CancelledError:
pass
async def main():
"""Main example runner"""
print("Choose example:")
print("1. Basic keepalive")
print("2. Advanced keepalive with monitoring")
choice = input("Enter choice (1 or 2): ").strip()
if choice == "1":
await basic_keepalive_example()
elif choice == "2":
await advanced_keepalive_example()
else:
print("Invalid choice")
if __name__ == "__main__":
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
asyncio.run(main())
## Best Practices
1. **Appropriate intervals**: Use 30-60 second ping intervals for most applications
2. **Health monitoring**: Track connection health and implement alerting
3. **Graceful recovery**: Handle connection failures with exponential backoff
4. **Resource cleanup**: Always stop keepalive tasks when shutting down
5. **Timeout handling**: Set reasonable timeouts for ping requests
6. **Logging**: Log keepalive events for debugging and monitoring
7. **Integration**: Integrate keepalive with your connection management system
# JavaScript/TypeScript
Complete JavaScript and TypeScript examples for integrating with NextBlock's gRPC API and QUIC transaction submission.
## Overview
These examples demonstrate how to:
* Establish secure gRPC connections with authentication
* Submit raw transaction bytes over QUIC for lower transport overhead
* Submit single and batched transactions with proper tipping
* Stream real-time tip floor data for dynamic tip optimization
* Maintain persistent connections with keepalive mechanisms
## Prerequisites
Install the required dependencies:
```bash
npm install @grpc/grpc-js @grpc/proto-loader
npm install @solana/web3.js @solana/spl-token
npm install typescript @types/node # For TypeScript support
# Generate JavaScript gRPC client from nextblock-proto
# See https://github.com/nextblock-ag/nextblock-proto for instructions
```
## Examples
### Core Examples
* [Connection](/api/examples/javascript/connection) - Establish gRPC connections with authentication
* [QUIC Transaction Submission](/api/examples/javascript/quic) - Send raw signed transaction bytes over QUIC
* [Submit Single Transaction](/api/examples/javascript/submit-single-transactions) - Send individual transactions with tips
* [Submit Batched Transactions](/api/examples/javascript/submit-batched-transactions) - Send atomic transaction bundles
* [Tip Floor Stream](/api/examples/javascript/tip-floor-stream) - Stream real-time tip floor data
* [Keepalive](/api/examples/javascript/keepalive) - Maintain persistent connections
## Quick Start
1. **Generate gRPC client** from [nextblock-proto](https://github.com/nextblock-ag/nextblock-proto)
2. **Install dependencies** using npm/yarn
3. **Set up environment** with API key and endpoint
4. **Start with connection example** to establish authenticated gRPC connection
5. **Use tip floor streaming** to optimize transaction tips dynamically
6. **Submit transactions** using single or batched submission methods
## Key Features
### TypeScript Support
All examples include full TypeScript type definitions for better development experience.
### Modern JavaScript
Uses modern ES6+ features including async/await, destructuring, and modules.
### Error Handling
Comprehensive error handling with retry logic and exponential backoff.
### Connection Management
Persistent connections with keepalive, health monitoring, and automatic recovery.
### Dynamic Tipping
Real-time tip optimization based on current network conditions from tip floor API.
## Best Practices
1. **Use TypeScript** - Leverage type safety for better code quality
2. **Enable TLS** - Always use secure connections in production
3. **Implement keepalive** - Maintain persistent connections for better performance
4. **Monitor tip floors** - Use streaming API for dynamic tip adjustment
5. **Handle errors gracefully** - Implement proper retry logic with exponential backoff
6. **Use environment variables** - Store sensitive configuration securely
7. **Choose appropriate endpoints** - Use the closest endpoint for better latency
8. **Bundle transactions** - Use batched submissions for atomic operations
# Connection
Establish a gRPC connection to NextBlock's API using Node.js and `@grpc/grpc-js`.
This page shows the connection pattern and authentication interceptor. Replace the placeholder generated client wiring with the client generated from [`nextblock-proto`](https://github.com/nextblock-ag/nextblock-proto).
## Prerequisites
Install the required dependencies:
```bash
npm install @grpc/grpc-js @grpc/proto-loader
npm install @solana/web3.js
npm install typescript @types/node # For TypeScript
```
Generate the gRPC client from proto specs:
```bash
# Clone the proto repository
git clone https://github.com/nextblock-ag/nextblock-proto
# Follow the JavaScript/TypeScript generation instructions in the repo
```
## Connection Setup
import * as grpc from '@grpc/grpc-js';
import * as protoLoader from '@grpc/proto-loader';
import { promisify } from 'util';
// Configuration interface
interface NextBlockConfig {
endpoint: string;
apiKey: string;
useTLS: boolean;
timeout: number;
keepaliveTimeMs: number;
keepaliveTimeoutMs: number;
}
// Default configuration
const defaultConfig: NextBlockConfig = {
endpoint: 'frankfurt.nextblock.io:443',
apiKey: process.env.NEXTBLOCK_API_KEY || '',
useTLS: true,
timeout: 30000,
keepaliveTimeMs: 60000, // 60 seconds
keepaliveTimeoutMs: 15000, // 15 seconds
};
// Authentication metadata interceptor
class AuthInterceptor {
private apiKey: string;
constructor(apiKey: string) {
this.apiKey = apiKey;
}
// Add authorization header to all requests
intercept(options: any, nextCall: any) {
return new grpc.InterceptingCall(nextCall(options), {
start: (metadata: grpc.Metadata, listener: any, next: any) => {
metadata.set('authorization', this.apiKey);
next(metadata, listener);
},
});
}
}
// Create gRPC channel with authentication
export async function createNextBlockChannel(config: NextBlockConfig): Promise<grpc.Channel> {
// Channel options for keepalive and performance
const channelOptions: grpc.ChannelOptions = {
'grpc.keepalive_time_ms': config.keepaliveTimeMs,
'grpc.keepalive_timeout_ms': config.keepaliveTimeoutMs,
'grpc.keepalive_permit_without_stream': 1,
'grpc.http2.max_pings_without_data': 0,
'grpc.http2.min_ping_interval_without_data_ms': 300000, // 5 minutes
};
// Create credentials
const credentials = config.useTLS
? grpc.credentials.createSsl()
: grpc.credentials.createInsecure();
// Create channel
const channel = new grpc.Channel(config.endpoint, credentials, channelOptions);
return channel;
}
// Create authenticated gRPC client
export async function createNextBlockClient(config: NextBlockConfig = defaultConfig) {
// Validate configuration
if (!config.apiKey) {
throw new Error('API key is required');
}
// Create channel
const channel = await createNextBlockChannel(config);
// Create authentication interceptor
const authInterceptor = new AuthInterceptor(config.apiKey);
// Load proto definition (replace with your generated client)
/* Example proto loading - replace with your actual proto
const packageDefinition = protoLoader.loadSync('path/to/your/nextblock.proto', {
keepCase: true,
longs: String,
enums: String,
defaults: true,
oneofs: true,
});
const protoDescriptor = grpc.loadPackageDefinition(packageDefinition) as any;
const ApiClient = protoDescriptor.nextblock.Api;
// Create client with interceptor
const client = new ApiClient(config.endpoint, credentials, {
interceptors: [authInterceptor.intercept.bind(authInterceptor)],
...channelOptions,
});
*/
console.log(`Connected to NextBlock at ${config.endpoint}`);
return {
channel,
// client, // Uncomment when you have the generated client
config,
};
}
// Connection manager with health checking
export class NextBlockConnectionManager {
private config: NextBlockConfig;
private channel?: grpc.Channel;
private client?: any; // Replace with your generated client type
private isConnected: boolean = false;
constructor(config: NextBlockConfig = defaultConfig) {
this.config = config;
}
// Establish connection
async connect(): Promise<boolean> {
try {
const connection = await createNextBlockClient(this.config);
this.channel = connection.channel;
// this.client = connection.client;
// Test the connection
await this.healthCheck();
this.isConnected = true;
console.log(`Successfully connected to NextBlock at ${this.config.endpoint}`);
return true;
} catch (error) {
console.error('Failed to connect to NextBlock:', error);
this.isConnected = false;
return false;
}
}
// Health check
async healthCheck(): Promise<boolean> {
if (!this.channel) {
return false;
}
return new Promise((resolve) => {
// Check channel state
const state = this.channel!.getConnectivityState(false);
if (state === grpc.connectivityState.READY) {
console.log('Connection health check passed');
resolve(true);
} else {
console.log(`Connection state: ${grpc.connectivityState[state]}`);
resolve(false);
}
});
}
// Disconnect
async disconnect(): Promise<void> {
if (this.channel) {
this.channel.close();
this.isConnected = false;
console.log('Disconnected from NextBlock');
}
}
// Getters
get connected(): boolean {
return this.isConnected;
}
get grpcClient(): any {
return this.client;
}
}
// Configuration from environment variables
export function configFromEnv(): NextBlockConfig {
return {
endpoint: process.env.NEXTBLOCK_ENDPOINT || 'frankfurt.nextblock.io:443',
apiKey: process.env.NEXTBLOCK_API_KEY || '',
useTLS: process.env.NEXTBLOCK_USE_TLS !== 'false',
timeout: parseInt(process.env.NEXTBLOCK_TIMEOUT || '30000'),
keepaliveTimeMs: parseInt(process.env.NEXTBLOCK_KEEPALIVE_TIME_MS || '60000'),
keepaliveTimeoutMs: parseInt(process.env.NEXTBLOCK_KEEPALIVE_TIMEOUT_MS || '15000'),
};
}
## Usage Examples
// Basic connection example
async function basicConnectionExample() {
const config: NextBlockConfig = {
endpoint: 'frankfurt.nextblock.io:443',
apiKey: '<your-api-key-here>',
useTLS: true,
timeout: 30000,
keepaliveTimeMs: 60000,
keepaliveTimeoutMs: 15000,
};
try {
const connection = await createNextBlockClient(config);
console.log('Connected to NextBlock!');
// Your API calls would go here
// const response = await connection.client.ping({});
// Clean up
connection.channel.close();
} catch (error) {
console.error('Connection failed:', error);
}
}
// Connection manager example (recommended)
async function connectionManagerExample() {
const config = configFromEnv();
const manager = new NextBlockConnectionManager(config);
try {
const connected = await manager.connect();
if (connected) {
console.log('Connection manager established connection');
// Use manager.grpcClient for API calls
// const response = await manager.grpcClient.ping({});
// Keep connection alive for your application
await new Promise(resolve => setTimeout(resolve, 5000));
} else {
console.error('Failed to establish connection');
}
} finally {
await manager.disconnect();
}
}
// Advanced connection with retry logic
class EnhancedConnectionManager extends NextBlockConnectionManager {
private maxRetries: number = 3;
private retryDelay: number = 5000; // 5 seconds
async connectWithRetry(): Promise<boolean> {
for (let attempt = 1; attempt <= this.maxRetries; attempt++) {
console.log(`Connection attempt ${attempt}/${this.maxRetries}`);
const success = await this.connect();
if (success) {
return true;
}
if (attempt < this.maxRetries) {
console.log(`Retrying in ${this.retryDelay}ms...`);
await new Promise(resolve => setTimeout(resolve, this.retryDelay));
this.retryDelay *= 2; // Exponential backoff
}
}
console.error('All connection attempts failed');
return false;
}
async callWithRetry<T>(
fn: () => Promise<T>,
maxRetries: number = 3
): Promise<T> {
let lastError: Error;
for (let attempt = 1; attempt <= maxRetries; attempt++) {
try {
return await fn();
} catch (error) {
lastError = error as Error;
// Don't retry on authentication errors
if (error && (error as any).code === grpc.status.UNAUTHENTICATED) {
throw error;
}
if (attempt < maxRetries) {
const delay = Math.min(1000 * Math.pow(2, attempt - 1), 30000);
console.log(`Attempt ${attempt} failed, retrying in ${delay}ms...`);
await new Promise(resolve => setTimeout(resolve, delay));
}
}
}
throw lastError!;
}
}
// Connection pool for high-throughput applications
class ConnectionPool {
private connections: NextBlockConnectionManager[] = [];
private currentIndex: number = 0;
private config: NextBlockConfig;
private poolSize: number;
constructor(config: NextBlockConfig, poolSize: number = 5) {
this.config = config;
this.poolSize = poolSize;
}
async initialize(): Promise<void> {
console.log(`Initializing connection pool with ${this.poolSize} connections...`);
const connectionPromises = Array.from({ length: this.poolSize }, async (_, i) => {
const manager = new NextBlockConnectionManager(this.config);
const connected = await manager.connect();
if (connected) {
this.connections.push(manager);
console.log(`Connection ${i + 1} established`);
} else {
console.warn(`Failed to create connection ${i + 1}`);
}
});
await Promise.all(connectionPromises);
console.log(`Connection pool initialized with ${this.connections.length} connections`);
}
getConnection(): NextBlockConnectionManager {
if (this.connections.length === 0) {
throw new Error('No available connections in pool');
}
const connection = this.connections[this.currentIndex];
this.currentIndex = (this.currentIndex + 1) % this.connections.length;
return connection;
}
async closeAll(): Promise<void> {
await Promise.all(this.connections.map(conn => conn.disconnect()));
this.connections = [];
console.log('All connections closed');
}
}
// Run examples
async function main() {
console.log('NextBlock Connection Examples');
// Example 1: Basic connection
console.log('\n1. Basic Connection:');
await basicConnectionExample();
// Example 2: Connection manager
console.log('\n2. Connection Manager:');
await connectionManagerExample();
// Example 3: Enhanced connection with retry
console.log('\n3. Enhanced Connection:');
const enhancedManager = new EnhancedConnectionManager(configFromEnv());
const connected = await enhancedManager.connectWithRetry();
if (connected) {
console.log('Enhanced connection successful');
await enhancedManager.disconnect();
}
}
// Export for use in other modules
export {
NextBlockConfig,
NextBlockConnectionManager,
EnhancedConnectionManager,
ConnectionPool,
configFromEnv,
};
if (require.main === module) {
main().catch(console.error);
}
## Available Endpoints
* **Frankfurt**: `frankfurt.nextblock.io:443` (Europe)
* **Amsterdam**: `amsterdam.nextblock.io:443` (Europe)
* **London**: `london.nextblock.io:443` (Europe)
* **Singapore**: `singapore.nextblock.io:443` (Asia)
* **Tokyo**: `tokyo.nextblock.io:443` (Asia)
* **New York**: `ny.nextblock.io:443` (US East)
* **Salt Lake City**: `slc.nextblock.io:443` (US West)
* **Dublin**: `dublin.nextblock.io:443` (Europe)
* **Vilnius**: `vilnius.nextblock.io:443` (Europe)
## Best Practices
1. **Use TypeScript**: Leverage type safety for better development experience
2. **Enable TLS**: Always use secure connections in production
3. **Implement keepalive**: Configure appropriate keepalive settings for persistent connections
4. **Handle errors gracefully**: Use retry logic with exponential backoff
5. **Validate configuration**: Check all required settings before connecting
6. **Use connection pooling**: For high-throughput applications
7. **Monitor connection health**: Implement regular health checks
8. **Environment variables**: Store sensitive configuration securely
9. **Close connections properly**: Always close connections when done
10. **Use interceptors**: Implement authentication and logging via gRPC interceptors
# QUIC Transaction Submission
Use QUIC when you already have signed Solana transaction bytes and want a low-overhead submission path from Node.js or TypeScript.
This example uses [`@matrixai/quic`](https://www.npmjs.com/package/@matrixai/quic) and follows the same flow as the other languages:
1. Connect to a regional endpoint with ALPN `nb-tx/1`
2. Authenticate once with your API key on a bidirectional stream
3. Reuse the connection and send each transaction on a unidirectional stream
## Example
```typescript
import { randomFillSync } from "node:crypto";
import { QUICClient } from "@matrixai/quic";
const AUTH_OK = 0x00;
const MAX_TX_SIZE = 1232;
class NextblockQuicClient {
constructor(private readonly client: QUICClient) {}
static async connect(serverAddr: string, apiKey: string): Promise {
const [host, portString] = serverAddr.split(":");
const port = Number(portString);
const client = await QUICClient.createQUICClient({
host,
port,
serverName: host,
crypto: {
ops: {
async randomBytes(data: ArrayBuffer): Promise {
randomFillSync(new Uint8Array(data));
},
},
},
config: {
applicationProtos: ["nb-tx/1"],
maxIdleTimeout: 60_000,
keepAliveIntervalTime: 15_000,
},
});
const authStream = client.connection.newStream("bidi");
const authWriter = authStream.writable.getWriter();
await authWriter.write(new TextEncoder().encode(apiKey));
await authWriter.close();
const authReader = authStream.readable.getReader();
const { value } = await authReader.read();
if (!value || value[0] !== AUTH_OK) {
await client.destroy();
throw new Error("Authentication rejected");
}
return new NextblockQuicClient(client);
}
async sendTransaction(rawTx: Uint8Array): Promise {
if (rawTx.byteLength > MAX_TX_SIZE) {
throw new Error(`Transaction too large: ${rawTx.byteLength}`);
}
const txStream = this.client.connection.newStream("uni");
const txWriter = txStream.writable.getWriter();
await txWriter.write(rawTx);
await txWriter.close();
}
async close(): Promise {
await this.client.destroy();
}
}
async function main(): Promise {
const apiKey = process.env.NEXTBLOCK_API_KEY;
if (!apiKey) {
throw new Error("Set NEXTBLOCK_API_KEY before running");
}
const client = await NextblockQuicClient.connect(
"amsterdam.nextblock.io:11100",
apiKey,
);
try {
// Replace this with the serialized bytes of your signed transaction.
// Example with @solana/web3.js: const rawTx = signedTransaction.serialize();
const rawTx = new Uint8Array([0, 1, 2, 3]);
await client.sendTransaction(rawTx);
console.log("transaction queued");
} finally {
await client.close();
}
}
void main();
```
## What To Replace
* Replace `amsterdam.nextblock.io:11100` with the region closest to you.
* Read the API key from `NEXTBLOCK_API_KEY` or your own config source.
* Replace `rawTx` with the serialized bytes of your signed transaction.
## Notes
* Send raw transaction bytes, not base64.
* Reuse one `NextblockQuicClient` for many sends.
* QUIC does not support the extra gRPC submission flags or atomic bundle submission.
* If you build transactions with `@solana/web3.js`, serialize the signed transaction first and pass the resulting `Uint8Array` into `sendTransaction()`.
See [QUIC Transaction Submission](/api/quic-transaction-submission) for the full endpoint list and protocol summary.
# Submit Single Transaction
Submit individual transactions to NextBlock using JavaScript/TypeScript with proper tipping and error handling.
If you want to send raw signed transaction bytes over QUIC instead of gRPC, see [QUIC Transaction Submission](/api/examples/javascript/quic).
This example shows transaction construction and the request shape you send to NextBlock. Replace the placeholder generated client types with the client generated from `nextblock-proto`.
import {
Connection,
Keypair,
PublicKey,
SystemProgram,
Transaction,
sendAndConfirmTransaction
} from '@solana/web3.js';
// NextBlock tip wallets for load balancing
const NEXTBLOCK_TIP_WALLETS = [
'NEXTbLoCkB51HpLBLojQfpyVAMorm3zzKg7w9NFdqid',
'nextBLoCkPMgmG8ZgJtABeScP35qLa2AMCNKntAP7Xc',
'NextbLoCkVtMGcV47JzewQdvBpLqT9TxQFozQkN98pE',
'NexTbLoCkWykbLuB1NkjXgFWkX9oAtcoagQegygXXA2',
'NeXTBLoCKs9F1y5PJS9CKrFNNLU1keHW71rfh7KgA1X',
'NexTBLockJYZ7QD7p2byrUa6df8ndV2WSd8GkbWqfbb',
'neXtBLock1LeC67jYd1QdAa32kbVeubsfPNTJC1V5At',
'nEXTBLockYgngeRmRrjDV31mGSekVPqZoMGhQEZtPVG',
];
// Submission options interface
interface SubmissionOptions {
skipPreflight?: boolean;
frontRunningProtection?: boolean;
revertOnFail?: boolean;
disableRetries?: boolean;
snipeTransaction?: boolean;
}
// Get random tip wallet for load balancing
function getRandomNextblockTipWallet(): PublicKey {
const randomIndex = Math.floor(Math.random() * NEXTBLOCK_TIP_WALLETS.length);
return new PublicKey(NEXTBLOCK_TIP_WALLETS[randomIndex]);
}
// Build transaction with tip
async function buildTransactionWithTip(
connection: Connection,
payer: Keypair,
tipAmount: number,
instructions: any[]
): Promise<Transaction> {
// Get recent blockhash
const { blockhash } = await connection.getLatestBlockhash('finalized');
// Create tip instruction (should be first)
const tipWallet = getRandomNextblockTipWallet();
const tipInstruction = SystemProgram.transfer({
fromPubkey: payer.publicKey,
toPubkey: tipWallet,
lamports: tipAmount,
});
// Create transaction with tip and user instructions
const transaction = new Transaction({
recentBlockhash: blockhash,
feePayer: payer.publicKey,
});
transaction.add(tipInstruction);
instructions.forEach(instruction => transaction.add(instruction));
// Sign transaction
transaction.sign(payer);
return transaction;
}
// Submit single transaction to NextBlock
async function submitSingleTransaction(
// nextblockClient: any, // Your generated gRPC client
connection: Connection,
signer: Keypair,
recipient: PublicKey,
transferAmount: number,
tipAmount: number,
options: SubmissionOptions = {}
): Promise<string> {
// Default options
const opts: SubmissionOptions = {
skipPreflight: true,
frontRunningProtection: false,
revertOnFail: false,
disableRetries: false,
snipeTransaction: false,
...options,
};
// Create transfer instruction
const transferInstruction = SystemProgram.transfer({
fromPubkey: signer.publicKey,
toPubkey: recipient,
lamports: transferAmount,
});
// Build transaction with tip
const transaction = await buildTransactionWithTip(
connection,
signer,
tipAmount,
[transferInstruction]
);
// Convert to base64 for submission
const serializedTx = transaction.serialize();
const base64Tx = serializedTx.toString('base64');
// Submit to NextBlock
/* Uncomment when you have the generated gRPC client
const request = {
transaction: { content: base64Tx },
skipPreFlight: opts.skipPreflight,
snipeTransaction: opts.snipeTransaction,
frontRunningProtection: opts.frontRunningProtection,
disableRetries: opts.disableRetries,
revertOnFail: opts.revertOnFail,
};
const response = await nextblockClient.postSubmitV2(request);
console.log('Transaction submitted successfully!');
console.log(`Signature: ${response.signature}`);
console.log(`UUID: ${response.uuid}`);
return response.signature;
*/
// Until your generated client is wired in, return the local signature
const signature = transaction.signature?.toString() || '';
console.log(`Local transaction built successfully: ${signature}`);
return signature;
}
// Submit with priority-based tip calculation
async function submitWithOptimalTip(
// nextblockClient: any,
connection: Connection,
signer: Keypair,
recipient: PublicKey,
transferAmount: number,
priorityLevel: 'conservative' | 'normal' | 'aggressive' | 'priority' = 'normal'
): Promise<string> {
// Tip amounts based on priority level
// Get tip amounts from current tip floor data
// Higher tips = higher priority
// Use tip floor streaming API to get current optimal tips
const tipAmount = await getOptimalTipFromTipFloor(priorityLevel);
return await submitSingleTransaction(
// nextblockClient,
connection,
signer,
recipient,
transferAmount,
tipAmount
);
}
// Transaction builder class for complex transactions
class TransactionBuilder {
private payer: Keypair;
private instructions: any[] = [];
private tipAmount: number = 1_000_000; // Default tip
constructor(payer: Keypair) {
this.payer = payer;
}
addInstruction(instruction: any): TransactionBuilder {
this.instructions.push(instruction);
return this;
}
setTipAmount(amount: number): TransactionBuilder {
this.tipAmount = amount;
return this;
}
async buildAndSubmit(
// nextblockClient: any,
connection: Connection,
options?: SubmissionOptions
): Promise<string> {
if (this.instructions.length === 0) {
throw new Error('No instructions added to transaction');
}
const transaction = await buildTransactionWithTip(
connection,
this.payer,
this.tipAmount,
this.instructions
);
// Convert and submit (similar to submitSingleTransaction)
const serializedTx = transaction.serialize();
const base64Tx = serializedTx.toString('base64');
const signature = transaction.signature?.toString() || '';
console.log(`Custom transaction built locally: ${signature}`);
return signature;
}
}
// Batch multiple single transactions (concurrent submission)
async function submitMultipleSingleTransactions(
// nextblockClient: any,
connection: Connection,
signer: Keypair,
transactionsData: Array<{
recipient: PublicKey;
transferAmount: number;
tipAmount: number;
}>
): Promise<string[]> {
// Create promises for all transactions
const transactionPromises = transactionsData.map(async ({ recipient, transferAmount, tipAmount }) => {
try {
return await submitSingleTransaction(
// nextblockClient,
connection,
signer,
recipient,
transferAmount,
tipAmount
);
} catch (error) {
console.error(`Transaction failed:`, error);
return null;
}
});
// Execute all transactions concurrently
const results = await Promise.allSettled(transactionPromises);
// Process results
const successfulSignatures: string[] = [];
results.forEach((result, index) => {
if (result.status === 'fulfilled' && result.value) {
successfulSignatures.push(result.value);
console.log(`Transaction ${index + 1} successful: ${result.value}`);
} else {
console.error(`Transaction ${index + 1} failed:`, result);
}
});
return successfulSignatures;
}
// Performance monitoring
class TransactionMetrics {
private submissions: number = 0;
private successes: number = 0;
private failures: number = 0;
private totalTime: number = 0;
private startTime?: number;
startSubmission(): void {
this.submissions++;
this.startTime = Date.now();
}
recordSuccess(): void {
if (this.startTime) {
this.totalTime += Date.now() - this.startTime;
this.successes++;
this.startTime = undefined;
}
}
recordFailure(): void {
if (this.startTime) {
this.totalTime += Date.now() - this.startTime;
this.failures++;
this.startTime = undefined;
}
}
getStats() {
return {
totalSubmissions: this.submissions,
successes: this.successes,
failures: this.failures,
successRate: (this.successes / Math.max(this.submissions, 1)) * 100,
averageTime: this.totalTime / Math.max(this.successes, 1),
};
}
}
// Submit with monitoring
async function submitWithMonitoring(
// nextblockClient: any,
connection: Connection,
signer: Keypair,
recipient: PublicKey,
transferAmount: number,
tipAmount: number,
metrics: TransactionMetrics
): Promise<string | null> {
metrics.startSubmission();
try {
const signature = await submitSingleTransaction(
// nextblockClient,
connection,
signer,
recipient,
transferAmount,
tipAmount
);
metrics.recordSuccess();
return signature;
} catch (error) {
metrics.recordFailure();
console.error('Transaction failed:', error);
return null;
}
}
## Usage Examples
async function main() {
// Initialize Solana connection
const connection = new Connection('https://api.mainnet-beta.solana.com', 'confirmed');
const signer = Keypair.generate(); // Use your actual keypair
const recipient = new PublicKey('<recipient-public-key>');
// Connect to NextBlock (see connection.md)
// const config = configFromEnv();
// const manager = new NextBlockConnectionManager(config);
// await manager.connect();
// const nextblockClient = manager.grpcClient;
try {
// Example 1: Basic transaction submission
console.log('1. Basic transaction submission:');
const signature1 = await submitSingleTransaction(
// nextblockClient,
connection,
signer,
recipient,
10_000, // Transfer 10,000 lamports
1_000_000 // Tip 1,000,000 lamports (0.001 SOL)
);
console.log(`Basic transaction: ${signature1}\n`);
// Example 2: Transaction with optimal tip
console.log('2. Transaction with optimal tip:');
const signature2 = await submitWithOptimalTip(
// nextblockClient,
connection,
signer,
recipient,
20_000, // Transfer 20,000 lamports
'aggressive' // Use aggressive tip strategy
);
console.log(`Optimally tipped transaction: ${signature2}\n`);
// Example 3: Custom transaction builder
console.log('3. Custom transaction builder:');
const builder = new TransactionBuilder(signer);
const signature3 = await builder
.addInstruction(SystemProgram.transfer({
fromPubkey: signer.publicKey,
toPubkey: recipient,
lamports: 30_000,
}))
.setTipAmount(1_500_000)
.buildAndSubmit(
// nextblockClient,
connection
);
console.log(`Custom built transaction: ${signature3}\n`);
// Example 4: Multiple concurrent transactions
console.log('4. Multiple concurrent transactions:');
const transactionsData = [
{ recipient, transferAmount: 5_000, tipAmount: 500_000 },
{ recipient, transferAmount: 7_500, tipAmount: 750_000 },
{ recipient, transferAmount: 12_500, tipAmount: 1_250_000 },
];
const signatures = await submitMultipleSingleTransactions(
// nextblockClient,
connection,
signer,
transactionsData
);
console.log(`Multiple transactions: ${signatures}\n`);
// Example 5: Transaction monitoring
console.log('5. Transaction with performance monitoring:');
const metrics = new TransactionMetrics();
for (let i = 0; i < 5; i++) {
await submitWithMonitoring(
// nextblockClient,
connection,
signer,
recipient,
1_000 * (i + 1), // Varying amounts
500_000, // Fixed tip
metrics
);
}
console.log('Performance metrics:', metrics.getStats());
} catch (error) {
console.error('Error in main:', error);
} finally {
// Clean up connection
// await manager.disconnect();
}
}
// Error handling with retry logic
async function submitWithRetry(
// nextblockClient: any,
connection: Connection,
signer: Keypair,
recipient: PublicKey,
transferAmount: number,
tipAmount: number,
maxRetries: number = 3
): Promise<string> {
let lastError: Error;
for (let attempt = 1; attempt <= maxRetries; attempt++) {
try {
return await submitSingleTransaction(
// nextblockClient,
connection,
signer,
recipient,
transferAmount,
tipAmount
);
} catch (error) {
lastError = error as Error;
if (attempt < maxRetries) {
const delay = Math.min(1000 * Math.pow(2, attempt - 1), 30000);
console.log(`Attempt ${attempt} failed, retrying in ${delay}ms...`);
await new Promise(resolve => setTimeout(resolve, delay));
}
}
}
throw lastError!;
}
if (require.main === module) {
main().catch(console.error);
}
export {
submitSingleTransaction,
submitWithOptimalTip,
TransactionBuilder,
submitMultipleSingleTransactions,
TransactionMetrics,
submitWithMonitoring,
submitWithRetry,
};
## Best Practices
1. **Always include tips**: NextBlock prioritizes transactions with appropriate tips
2. **Use random tip wallets**: Distribute load across multiple tip addresses
3. **Monitor tip floors**: Adjust tips based on current network conditions
4. **Handle errors gracefully**: Implement retry logic with exponential backoff
5. **Validate inputs**: Always validate public keys and amounts before submission
6. **Use TypeScript**: Leverage type safety for better development experience
7. **Monitor performance**: Track success rates and response times
8. **Choose appropriate RPC endpoints**: Use reliable RPC providers for blockhash retrieval
# Submit Batched Transactions
Submit 2-4 transactions as an atomic bundle to NextBlock using JavaScript/TypeScript.
This example shows bundle construction and the request shape you send to NextBlock. Replace the placeholder generated client types with the client generated from `nextblock-proto`.
import {
Connection,
Keypair,
PublicKey,
SystemProgram,
Transaction
} from '@solana/web3.js';
// Transaction bundle builder
class TransactionBundle {
private transactions: Transaction[] = [];
private readonly maxSize: number = 4;
private readonly minSize: number = 2;
addTransaction(transaction: Transaction): TransactionBundle {
if (this.transactions.length >= this.maxSize) {
throw new Error(`Bundle cannot contain more than ${this.maxSize} transactions`);
}
this.transactions.push(transaction);
return this;
}
validate(): boolean {
if (this.transactions.length < this.minSize) {
throw new Error(`Bundle must contain at least ${this.minSize} transactions`);
}
if (this.transactions.length > this.maxSize) {
throw new Error(`Bundle cannot contain more than ${this.maxSize} transactions`);
}
// Check for duplicate signatures
const signatures = new Set();
for (const tx of this.transactions) {
const sig = tx.signature?.toString();
if (sig && signatures.has(sig)) {
throw new Error('Duplicate transaction signatures in bundle');
}
if (sig) signatures.add(sig);
}
return true;
}
toBase64Transactions(): string[] {
this.validate();
return this.transactions.map(tx => {
const serialized = tx.serialize();
return serialized.toString('base64');
});
}
getSignatures(): string[] {
return this.transactions
.map(tx => tx.signature?.toString())
.filter((sig): sig is string => sig !== undefined);
}
get size(): number {
return this.transactions.length;
}
}
// Build single transaction with tip
async function buildTransactionWithTip(
connection: Connection,
payer: Keypair,
tipAmount: number,
instructions: any[]
): Promise<Transaction> {
const { blockhash } = await connection.getLatestBlockhash('finalized');
// Random tip wallet for load balancing
const tipWallets = [
'NEXTbLoCkB51HpLBLojQfpyVAMorm3zzKg7w9NFdqid',
'nextBLoCkPMgmG8ZgJtABeScP35qLa2AMCNKntAP7Xc',
'NextbLoCkVtMGcV47JzewQdvBpLqT9TxQFozQkN98pE',
'NexTbLoCkWykbLuB1NkjXgFWkX9oAtcoagQegygXXA2',
];
const tipWallet = new PublicKey(tipWallets[Math.floor(Math.random() * tipWallets.length)]);
// Create tip instruction (should be first)
const tipInstruction = SystemProgram.transfer({
fromPubkey: payer.publicKey,
toPubkey: tipWallet,
lamports: tipAmount,
});
// Create and sign transaction
const transaction = new Transaction({
recentBlockhash: blockhash,
feePayer: payer.publicKey,
});
transaction.add(tipInstruction);
instructions.forEach(instruction => transaction.add(instruction));
transaction.sign(payer);
return transaction;
}
// Submit batched transactions
async function submitBatchedTransactions(
// nextblockClient: any, // Your generated gRPC client
connection: Connection,
signer: Keypair,
transactionSpecs: Array<{
instructions: any[];
tipAmount: number;
}>
): Promise<string> {
// Build transaction bundle
const bundle = new TransactionBundle();
for (const { instructions, tipAmount } of transactionSpecs) {
const transaction = await buildTransactionWithTip(
connection,
signer,
tipAmount,
instructions
);
bundle.addTransaction(transaction);
}
// Get base64 transactions for submission
const base64Transactions = bundle.toBase64Transactions();
const signatures = bundle.getSignatures();
// Log transaction signatures
signatures.forEach((sig, i) => {
console.log(`Transaction ${i + 1} signature: ${sig}`);
});
// Submit bundle to NextBlock
/* Uncomment when you have the generated gRPC client
const entries = base64Transactions.map(base64Tx => ({
transaction: { content: base64Tx }
}));
const request = { entries };
const response = await nextblockClient.postSubmitBatchV2(request);
console.log('Batch submitted successfully!');
console.log(`Bundle signature: ${response.signature}`);
return response.signature;
*/
// Until your generated client is wired in, return a placeholder bundle signature
const bundleSignature = '';
console.log(`Local bundle prepared with ${bundle.size} transactions.`);
return bundleSignature;
}
// Build arbitrage bundle
async function buildArbitrageBundle(
// nextblockClient: any,
connection: Connection,
signer: Keypair,
dexAAddress: PublicKey,
dexBAddress: PublicKey,
tradeAmount: number
): Promise<string> {
// Transaction 1: Buy on DEX A
const buyInstructions = [
// Add your DEX-specific buy instructions here
SystemProgram.transfer({
fromPubkey: signer.publicKey,
toPubkey: dexAAddress,
lamports: tradeAmount,
})
];
// Transaction 2: Sell on DEX B
const sellInstructions = [
// Add your DEX-specific sell instructions here
SystemProgram.transfer({
fromPubkey: signer.publicKey,
toPubkey: dexBAddress,
lamports: tradeAmount,
})
];
const transactionSpecs = [
{ instructions: buyInstructions, tipAmount: 2_000_000 }, // Higher tip for arbitrage
{ instructions: sellInstructions, tipAmount: 2_000_000 }, // Higher tip for arbitrage
];
return await submitBatchedTransactions(
// nextblockClient,
connection,
signer,
transactionSpecs
);
}
// Build complex DeFi operation bundle
async function buildDeFiOperationBundle(
// nextblockClient: any,
connection: Connection,
signer: Keypair
): Promise<string> {
// Transaction 1: Setup - Create token accounts
const setupInstructions = [
// Add token account creation instructions
SystemProgram.transfer({
fromPubkey: signer.publicKey,
toPubkey: new PublicKey('<token-program-address>'),
lamports: 1_000_000 // Rent exemption
})
];
// Transaction 2: Main operation - Execute swap
const swapInstructions = [
// Add swap instructions (Jupiter, Raydium, etc.)
SystemProgram.transfer({
fromPubkey: signer.publicKey,
toPubkey: new PublicKey('<swap-program-address>'),
lamports: 0 // No SOL transfer for swap
})
];
// Transaction 3: Cleanup - Stake or provide liquidity
const stakeInstructions = [
// Add staking/liquidity instructions
SystemProgram.transfer({
fromPubkey: signer.publicKey,
toPubkey: new PublicKey('<stake-pool-address>'),
lamports: 0
})
];
const transactionSpecs = [
{ instructions: setupInstructions, tipAmount: 500_000 }, // Setup tip
{ instructions: swapInstructions, tipAmount: 1_500_000 }, // Main operation tip
{ instructions: stakeInstructions, tipAmount: 750_000 }, // Cleanup tip
];
return await submitBatchedTransactions(
// nextblockClient,
connection,
signer,
transactionSpecs
);
}
// Bundle optimizer
class BundleOptimizer {
private tipMultipliers = {
setup: 0.5, // Lower priority
main: 1.5, // Higher priority
cleanup: 0.75, // Medium priority
arbitrage: 2.0, // Highest priority
};
optimizeTips(baseTip: number, transactionTypes: string[]): number[] {
return transactionTypes.map(txType => {
const multiplier = this.tipMultipliers[txType as keyof typeof this.tipMultipliers] || 1.0;
return Math.floor(baseTip * multiplier);
});
}
reorderTransactions<T>(
transactions: T[],
transactionTypes: string[]
): { transactions: T[]; types: string[] } {
// Priority order: setup -> main -> cleanup
const priorityOrder = { setup: 1, main: 2, cleanup: 3, arbitrage: 0 };
const pairs = transactions.map((tx, i) => ({
transaction: tx,
type: transactionTypes[i],
priority: priorityOrder[transactionTypes[i] as keyof typeof priorityOrder] || 2
}));
pairs.sort((a, b) => a.priority - b.priority);
return {
transactions: pairs.map(p => p.transaction),
types: pairs.map(p => p.type)
};
}
}
// Bundle status tracking
interface BundleStatus {
bundleId: string;
transactionCount: number;
submittedAt: number;
signatures: string[];
status: 'pending' | 'confirmed' | 'failed';
}
class BundleTracker {
private bundles: Map<string, BundleStatus> = new Map();
trackBundle(bundleStatus: BundleStatus): void {
this.bundles.set(bundleStatus.bundleId, bundleStatus);
}
async checkBundleStatus(
bundleId: string,
connection: Connection
): Promise<BundleStatus | null> {
const bundleStatus = this.bundles.get(bundleId);
if (!bundleStatus) return null;
// Check if all transactions are confirmed
let confirmedCount = 0;
for (const signature of bundleStatus.signatures) {
try {
const status = await connection.getSignatureStatus(signature);
if (status.value?.confirmationStatus) {
confirmedCount++;
}
} catch (error) {
console.error(`Failed to check signature ${signature}:`, error);
}
}
// Update bundle status
if (confirmedCount === bundleStatus.transactionCount) {
bundleStatus.status = 'confirmed';
} else if (Date.now() - bundleStatus.submittedAt > 60000) { // Timeout after 60 seconds
bundleStatus.status = 'failed';
}
return bundleStatus;
}
getBundleStatus(bundleId: string): BundleStatus | null {
return this.bundles.get(bundleId) || null;
}
}
## Usage Examples
async function main() {
// Initialize Solana connection
const connection = new Connection('https://api.mainnet-beta.solana.com', 'confirmed');
const signer = Keypair.generate(); // Use your actual keypair
// Connect to NextBlock (see connection.md)
// const config = configFromEnv();
// const manager = new NextBlockConnectionManager(config);
// await manager.connect();
// const nextblockClient = manager.grpcClient;
try {
// Example 1: Basic batch submission
console.log('1. Basic batch submission:');
const transactionSpecs = [
// Setup transaction
{
instructions: [SystemProgram.transfer({
fromPubkey: signer.publicKey,
toPubkey: new PublicKey('<recipient1>'),
lamports: 100_000,
})],
tipAmount: 500_000, // 0.0005 SOL tip
},
// Main transaction
{
instructions: [SystemProgram.transfer({
fromPubkey: signer.publicKey,
toPubkey: new PublicKey('<recipient2>'),
lamports: 200_000,
})],
tipAmount: 1_000_000, // 0.001 SOL tip
},
// Cleanup transaction
{
instructions: [SystemProgram.transfer({
fromPubkey: signer.publicKey,
toPubkey: new PublicKey('<recipient3>'),
lamports: 50_000,
})],
tipAmount: 500_000, // 0.0005 SOL tip
},
];
const bundleSignature = await submitBatchedTransactions(
// nextblockClient,
connection,
signer,
transactionSpecs
);
console.log(`Basic batch: ${bundleSignature}\n`);
// Example 2: Arbitrage bundle
console.log('2. Arbitrage bundle:');
const arbSignature = await buildArbitrageBundle(
// nextblockClient,
connection,
signer,
new PublicKey('<dex-a-address>'),
new PublicKey('<dex-b-address>'),
1_000_000 // 0.001 SOL trade
);
console.log(`Arbitrage bundle: ${arbSignature}\n`);
// Example 3: Optimized DeFi bundle
console.log('3. DeFi operation bundle:');
const defiSignature = await buildDeFiOperationBundle(
// nextblockClient,
connection,
signer
);
console.log(`DeFi bundle: ${defiSignature}\n`);
// Example 4: Bundle optimization
console.log('4. Bundle optimization:');
const optimizer = new BundleOptimizer();
const baseTip = 1_000_000;
const transactionTypes = ['setup', 'main', 'cleanup'];
const optimizedTips = optimizer.optimizeTips(baseTip, transactionTypes);
console.log(`Optimized tips: ${optimizedTips}`);
// Example 5: Bundle tracking
console.log('5. Bundle tracking:');
const tracker = new BundleTracker();
const bundleStatus: BundleStatus = {
bundleId: 'bundle-123',
transactionCount: 3,
submittedAt: Date.now(),
signatures: ['sig1', 'sig2', 'sig3'],
status: 'pending'
};
tracker.trackBundle(bundleStatus);
// Check status after some time
setTimeout(async () => {
const status = await tracker.checkBundleStatus('bundle-123', connection);
console.log('Bundle status:', status);
}, 5000);
} catch (error) {
console.error('Error in main:', error);
} finally {
// Clean up connection
// await manager.disconnect();
}
}
if (require.main === module) {
main().catch(console.error);
}
export {
TransactionBundle,
submitBatchedTransactions,
buildArbitrageBundle,
buildDeFiOperationBundle,
BundleOptimizer,
BundleTracker,
};
## Best Practices
1. **Bundle size limits**: Keep bundles between 2-4 transactions for optimal success rates
2. **Transaction ordering**: Setup → Main operations → Cleanup
3. **Progressive tipping**: Use higher tips for more critical transactions
4. **Error handling**: Validate bundles before submission
5. **Performance monitoring**: Track bundle success rates and timing
6. **Tip optimization**: Adjust tips based on transaction importance and network conditions
7. **Use TypeScript**: Leverage type safety for complex bundle operations
# Tip Floor Stream
Stream real-time tip floor data from NextBlock using JavaScript/TypeScript.
// Tip floor data interface
interface TipFloorData {
time: string;
landed_tips_25th_percentile: number;
landed_tips_50th_percentile: number;
landed_tips_75th_percentile: number;
landed_tips_95th_percentile: number;
landed_tips_99th_percentile: number;
ema_landed_tips_50th_percentile: number;
}
// Tip strategy management
class TipStrategy {
conservativeTip: number = 0; // Will be set from tip floor data
normalTip: number = 0; // Will be set from tip floor data
aggressiveTip: number = 0; // Will be set from tip floor data
priorityTip: number = 0; // Will be set from tip floor data
lastUpdated?: Date;
updateFromTipFloor(tipFloor: TipFloorData): void {
this.conservativeTip = this.solToLamports(tipFloor.landed_tips_25th_percentile);
this.normalTip = this.solToLamports(tipFloor.landed_tips_50th_percentile);
this.aggressiveTip = this.solToLamports(tipFloor.landed_tips_75th_percentile);
this.priorityTip = this.solToLamports(tipFloor.landed_tips_95th_percentile);
this.lastUpdated = new Date();
}
private solToLamports(sol: number): number {
return Math.floor(sol * 1_000_000_000);
}
getTipForPriority(priority: 'conservative' | 'normal' | 'aggressive' | 'priority'): number {
switch (priority) {
case 'conservative': return this.conservativeTip;
case 'normal': return this.normalTip;
case 'aggressive': return this.aggressiveTip;
case 'priority': return this.priorityTip;
default: return this.normalTip;
}
}
}
// Global tip strategy instance
const globalTipStrategy = new TipStrategy();
// Stream tip floor data
async function streamTipFloor(
// nextblockClient: any, // Your generated gRPC client
updateFrequency: string = '1m',
callback?: (tipFloor: TipFloorData) => Promise<void> | void
): Promise<void> {
console.log(`Starting tip floor stream with frequency: ${updateFrequency}`);
/* Uncomment when you have the generated gRPC client
try {
const request = { updateFrequency };
const stream = nextblockClient.streamTipFloor(request);
console.log('Streaming tip floor data:');
for await (const tipFloorResponse of stream) {
try {
const tipFloor: TipFloorData = {
time: tipFloorResponse.time,
landed_tips_25th_percentile: tipFloorResponse.landed_tips_25th_percentile,
landed_tips_50th_percentile: tipFloorResponse.landed_tips_50th_percentile,
landed_tips_75th_percentile: tipFloorResponse.landed_tips_75th_percentile,
landed_tips_95th_percentile: tipFloorResponse.landed_tips_95th_percentile,
landed_tips_99th_percentile: tipFloorResponse.landed_tips_99th_percentile,
ema_landed_tips_50th_percentile: tipFloorResponse.ema_landed_tips_50th_percentile,
};
console.log('Received tip floor update:');
console.log(` Time: ${tipFloor.time}`);
console.log(` 25th percentile: ${tipFloor.landed_tips_25th_percentile.toFixed(6)} SOL`);
console.log(` 50th percentile: ${tipFloor.landed_tips_50th_percentile.toFixed(6)} SOL`);
console.log(` 75th percentile: ${tipFloor.landed_tips_75th_percentile.toFixed(6)} SOL`);
console.log(` 95th percentile: ${tipFloor.landed_tips_95th_percentile.toFixed(6)} SOL`);
console.log(` EMA 50th percentile: ${tipFloor.ema_landed_tips_50th_percentile.toFixed(6)} SOL`);
console.log(' ---');
// Update global tip strategy
await processTipFloorUpdate(tipFloor);
// Call custom callback if provided
if (callback) {
await callback(tipFloor);
}
} catch (error) {
console.error('Error processing tip floor update:', error);
}
}
} catch (error) {
console.error('Tip floor stream error:', error);
// Implement reconnection logic here
}
*/
// Mock streaming for demonstration
console.log('Mock tip floor streaming started...');
const mockStream = setInterval(async () => {
// Generate mock tip floor data
const mockTipFloor: TipFloorData = {
time: new Date().toISOString(),
landed_tips_25th_percentile: 0.0011,
landed_tips_50th_percentile: 0.005000001,
landed_tips_75th_percentile: 0.01555,
landed_tips_95th_percentile: 0.09339195639999975,
landed_tips_99th_percentile: 0.4846427910400001,
ema_landed_tips_50th_percentile: 0.005989477267191758,
};
console.log('Mock tip floor update:', mockTipFloor);
await processTipFloorUpdate(mockTipFloor);
if (callback) {
await callback(mockTipFloor);
}
}, 60000); // Update every minute
// Return a promise that never resolves (keeps streaming)
return new Promise(() => {
// Keep the interval running
process.on('SIGINT', () => {
clearInterval(mockStream);
console.log('Tip floor streaming stopped');
process.exit(0);
});
});
}
// Process tip floor updates
async function processTipFloorUpdate(tipFloor: TipFloorData): Promise<void> {
// Update global strategy
globalTipStrategy.updateFromTipFloor(tipFloor);
console.log('Updated tip strategy:');
console.log(` Conservative: ${globalTipStrategy.conservativeTip} lamports`);
console.log(` Normal: ${globalTipStrategy.normalTip} lamports`);
console.log(` Aggressive: ${globalTipStrategy.aggressiveTip} lamports`);
console.log(` Priority: ${globalTipStrategy.priorityTip} lamports`);
// Store historical data
await storeTipFloorData(tipFloor);
// Trigger any pending transactions
await triggerPendingTransactions();
}
// Historical data management
class TipFloorHistory {
private data: TipFloorData[] = [];
private readonly maxSize: number;
constructor(maxSize: number = 1000) {
this.maxSize = maxSize;
}
add(tipFloor: TipFloorData): void {
if (this.data.length >= this.maxSize) {
this.data.shift(); // Remove oldest
}
this.data.push(tipFloor);
}
getTrend(percentile: '25th' | '50th' | '75th' | '95th' = '50th', window: number = 10): number {
if (this.data.length < 2) return 0;
const recentData = this.data.slice(-window);
if (recentData.length < 2) return 0;
const key = `landed_tips_${percentile}_percentile` as keyof TipFloorData;
const startValue = recentData[0][key] as number;
const endValue = recentData[recentData.length - 1][key] as number;
return endValue - startValue;
}
getAverage(percentile: '25th' | '50th' | '75th' | '95th' = '50th', window: number = 10): number {
if (this.data.length === 0) return 0;
const recentData = this.data.slice(-window);
const key = `landed_tips_${percentile}_percentile` as keyof TipFloorData;
const values = recentData.map(d => d[key] as number);
return values.reduce((sum, val) => sum + val, 0) / values.length;
}
get length(): number {
return this.data.length;
}
}
// Global history tracker
const tipFloorHistory = new TipFloorHistory();
// Smart tip calculation with trend analysis
async function getSmartTip(
basePriority: 'conservative' | 'normal' | 'aggressive' | 'priority' = 'normal',
considerTrend: boolean = true
): Promise<number> {
const baseTip = globalTipStrategy.getTipForPriority(basePriority);
if (!considerTrend || tipFloorHistory.length < 2) {
return baseTip;
}
// Analyze trend
const trend = tipFloorHistory.getTrend('50th', 5);
// Adjust tip based on trend
let adjustmentFactor = 1.0;
if (trend > 0.001) {
adjustmentFactor = 1.2;
console.log(`Tips trending up (+${trend.toFixed(6)}), increasing tip by 20%`);
} else if (trend < -0.001) {
adjustmentFactor = 0.9;
console.log(`Tips trending down (${trend.toFixed(6)}), decreasing tip by 10%`);
} else {
console.log(`Tips stable (${trend.toFixed(6)}), no adjustment`);
}
const smartTip = Math.floor(baseTip * adjustmentFactor);
return Math.max(smartTip, 100_000); // Minimum tip of 0.0001 SOL
}
// Store tip floor data
async function storeTipFloorData(tipFloor: TipFloorData): Promise<void> {
// Add to history
tipFloorHistory.add(tipFloor);
// Optionally save to file
const fs = require('fs').promises;
const filename = `tip_data_${new Date().toISOString().split('T')[0]}.jsonl`;
try {
await fs.appendFile(filename, JSON.stringify(tipFloor) + '\n');
} catch (error) {
console.error('Failed to store tip floor data:', error);
}
}
// Trigger pending transactions
async function triggerPendingTransactions(): Promise<void> {
console.log('Checking for pending transactions to trigger...');
// Implementation would check your pending transaction queue
// and submit them with updated tip amounts
}
// Get current optimal tips
function getCurrentOptimalTips(): {
conservative: number;
normal: number;
aggressive: number;
priority: number;
} {
return {
conservative: globalTipStrategy.getTipForPriority('conservative'),
normal: globalTipStrategy.getTipForPriority('normal'),
aggressive: globalTipStrategy.getTipForPriority('aggressive'),
priority: globalTipStrategy.getTipForPriority('priority'),
};
}
## Usage Examples
async function tipFloorExample() {
// Connect to NextBlock (see connection.md)
// const config = configFromEnv();
// const manager = new NextBlockConnectionManager(config);
// await manager.connect();
// const nextblockClient = manager.grpcClient;
// Custom callback for tip floor updates
const onTipFloorUpdate = async (tipFloor: TipFloorData) => {
console.log(`Custom handler: Received update at ${tipFloor.time}`);
// Example: Trigger high-priority transactions when tips are low
if (tipFloor.landed_tips_50th_percentile < 0.002) { // Less than 0.002 SOL
console.log('Tips are low - good time for high-priority transactions!');
// await submitPriorityTransactions();
}
};
// Start streaming in background
const streamPromise = streamTipFloor(
// nextblockClient,
'1m',
onTipFloorUpdate
);
// Example usage of dynamic tips
setTimeout(async () => {
// Get current optimal tips
const currentTips = getCurrentOptimalTips();
console.log('Current optimal tips:', currentTips);
// Get smart tip with trend analysis
const smartTip = await getSmartTip('normal', true);
console.log(`Smart tip: ${smartTip} lamports`);
// Example: Use tips in transaction submission
// await submitTransactionWithTip(smartTip);
}, 5000); // Wait for initial data
// Keep streaming
try {
await streamPromise;
} catch (error) {
console.error('Streaming error:', error);
}
}
// Advanced tip management example
async function advancedTipManagement() {
// Monitor tip trends and adjust strategy
const monitorTrends = setInterval(async () => {
if (tipFloorHistory.length >= 10) {
const trend5min = tipFloorHistory.getTrend('50th', 5);
const trend10min = tipFloorHistory.getTrend('50th', 10);
const average = tipFloorHistory.getAverage('50th', 10);
console.log('Tip Analysis:');
console.log(` 5-min trend: ${trend5min.toFixed(6)} SOL`);
console.log(` 10-min trend: ${trend10min.toFixed(6)} SOL`);
console.log(` 10-min average: ${average.toFixed(6)} SOL`);
// Adjust strategy based on trends
if (trend5min > 0.002 && trend10min > 0.001) {
console.log('Strong upward trend detected - consider higher tips');
} else if (trend5min < -0.002 && trend10min < -0.001) {
console.log('Strong downward trend detected - can use lower tips');
}
}
}, 300000); // Check every 5 minutes
// Clean up on exit
process.on('SIGINT', () => {
clearInterval(monitorTrends);
console.log('Advanced tip management stopped');
process.exit(0);
});
}
// Main example runner
async function main() {
console.log('NextBlock Tip Floor Streaming Examples');
try {
// Start tip floor streaming
console.log('Starting tip floor streaming...');
// Run both examples concurrently
await Promise.all([
tipFloorExample(),
advancedTipManagement()
]);
} catch (error) {
console.error('Error in main:', error);
}
}
if (require.main === module) {
main().catch(console.error);
}
export {
TipFloorData,
TipStrategy,
TipFloorHistory,
streamTipFloor,
getSmartTip,
getCurrentOptimalTips,
globalTipStrategy,
};
# Keepalive
Maintain persistent gRPC connections to NextBlock using JavaScript/TypeScript.
// Connection health tracking
interface ConnectionHealth {
isHealthy: boolean;
lastSuccessfulPing: Date;
consecutiveFailures: number;
totalPingsSent: number;
totalPingsSuccessful: number;
averagePingTime: number;
}
// Keepalive configuration
interface KeepaliveConfig {
pingInterval: number; // milliseconds
maxConsecutiveFailures: number;
reconnectDelay: number; // milliseconds
healthCheckEnabled: boolean;
timeout: number; // milliseconds
}
// Default keepalive configuration
const defaultKeepaliveConfig: KeepaliveConfig = {
pingInterval: 60000, // 60 seconds
maxConsecutiveFailures: 3,
reconnectDelay: 5000, // 5 seconds
healthCheckEnabled: true,
timeout: 15000, // 15 seconds
};
// Basic keepalive implementation
async function startKeepaliveTask(
// nextblockClient: any, // Your generated gRPC client
config: KeepaliveConfig = defaultKeepaliveConfig
): Promise<void> {
console.log(`Starting keepalive task with ${config.pingInterval}ms interval`);
const keepaliveInterval = setInterval(async () => {
try {
const startTime = Date.now();
/* Uncomment when you have the generated gRPC client
try {
await Promise.race([
nextblockClient.ping({}),
new Promise((_, reject) =>
setTimeout(() => reject(new Error('Timeout')), config.timeout)
)
]);
const pingTime = Date.now() - startTime;
console.log(`Keepalive ping successful (${pingTime}ms) at ${new Date().toTimeString()}`);
} catch (error) {
console.error('Keepalive ping failed:', error);
// Optionally implement reconnection logic
clearInterval(keepaliveInterval);
return;
}
*/
// Mock ping for demonstration
await new Promise(resolve => setTimeout(resolve, 50 + Math.random() * 100)); // 50-150ms delay
const pingTime = Date.now() - startTime;
console.log(`Mock keepalive ping successful (${pingTime}ms) at ${new Date().toTimeString()}`);
} catch (error) {
console.error('Keepalive task error:', error);
clearInterval(keepaliveInterval);
}
}, config.pingInterval);
// Handle graceful shutdown
process.on('SIGINT', () => {
clearInterval(keepaliveInterval);
console.log('Keepalive task stopped');
process.exit(0);
});
// Return a promise that never resolves (keeps running)
return new Promise(() => {});
}
// Advanced keepalive manager
class KeepaliveManager {
private config: KeepaliveConfig;
private health: ConnectionHealth;
private keepaliveInterval?: NodeJS.Timeout;
private isRunning: boolean = false;
constructor(
// private nextblockClient: any, // Your generated gRPC client
config: KeepaliveConfig = defaultKeepaliveConfig
) {
this.config = config;
this.health = {
isHealthy: true,
lastSuccessfulPing: new Date(),
consecutiveFailures: 0,
totalPingsSent: 0,
totalPingsSuccessful: 0,
averagePingTime: 0,
};
}
async start(): Promise<void> {
if (this.isRunning) return;
this.isRunning = true;
this.keepaliveInterval = setInterval(
() => this.sendPing(),
this.config.pingInterval
);
console.log('Keepalive manager started');
}
async stop(): Promise<void> {
this.isRunning = false;
if (this.keepaliveInterval) {
clearInterval(this.keepaliveInterval);
this.keepaliveInterval = undefined;
}
console.log('Keepalive manager stopped');
}
private async sendPing(): Promise<void> {
const startTime = Date.now();
try {
/* Uncomment when you have the generated gRPC client
await Promise.race([
this.nextblockClient.ping({}),
new Promise((_, reject) =>
setTimeout(() => reject(new Error('Timeout')), this.config.timeout)
)
]);
*/
// Mock ping delay
await new Promise(resolve =>
setTimeout(resolve, 50 + Math.random() * 100)
);
const pingTime = Date.now() - startTime;
this.updateHealthSuccess(pingTime);
console.log(
`Keepalive ping successful (${pingTime}ms) - ` +
`Health: ${this.getSuccessRate().toFixed(1)}%`
);
} catch (error) {
this.updateHealthFailure();
console.error(`Keepalive ping failed: ${error}`);
if (!this.health.isHealthy) {
await this.handleConnectionRecovery();
}
}
}
private updateHealthSuccess(pingTime: number): void {
this.health.isHealthy = true;
this.health.lastSuccessfulPing = new Date();
this.health.consecutiveFailures = 0;
this.health.totalPingsSent++;
this.health.totalPingsSuccessful++;
// Update average ping time
if (this.health.totalPingsSuccessful === 1) {
this.health.averagePingTime = pingTime;
} else {
this.health.averagePingTime = (
(this.health.averagePingTime * (this.health.totalPingsSuccessful - 1) + pingTime) /
this.health.totalPingsSuccessful
);
}
}
private updateHealthFailure(): void {
this.health.consecutiveFailures++;
this.health.totalPingsSent++;
if (this.health.consecutiveFailures >= this.config.maxConsecutiveFailures) {
this.health.isHealthy = false;
console.warn(
`Connection marked as unhealthy after ${this.health.consecutiveFailures} consecutive failures`
);
}
}
private async handleConnectionRecovery(): Promise<void> {
console.warn('Connection unhealthy, attempting recovery...');
// Wait before attempting recovery
await new Promise(resolve => setTimeout(resolve, this.config.reconnectDelay));
try {
// Implement connection recovery logic here
console.log('Connection recovery attempted');
// Reset some health metrics on successful recovery
// this.health.consecutiveFailures = 0;
} catch (error) {
console.error('Connection recovery failed:', error);
}
}
getHealth(): ConnectionHealth {
return { ...this.health };
}
isHealthy(): boolean {
return this.health.isHealthy;
}
private getSuccessRate(): number {
if (this.health.totalPingsSent === 0) return 100;
return (this.health.totalPingsSuccessful / this.health.totalPingsSent) * 100;
}
}
// Connection manager with integrated keepalive
class ConnectionManagerWithKeepalive {
private nextblockConfig: any;
private keepaliveConfig: KeepaliveConfig;
private keepaliveManager?: KeepaliveManager;
// private client?: any;
private isConnected: boolean = false;
constructor(
nextblockConfig: any,
keepaliveConfig: KeepaliveConfig = defaultKeepaliveConfig
) {
this.nextblockConfig = nextblockConfig;
this.keepaliveConfig = keepaliveConfig;
}
async connect(): Promise<boolean> {
try {
// Create connection (see connection.md)
// const connection = await createNextBlockClient(this.nextblockConfig);
// this.client = connection.client;
// Test connection
// await this.client.ping({});
this.isConnected = true;
console.log('Successfully connected to NextBlock');
// Start keepalive
this.keepaliveManager = new KeepaliveManager(
// this.client,
this.keepaliveConfig
);
await this.keepaliveManager.start();
return true;
} catch (error) {
console.error('Failed to connect:', error);
this.isConnected = false;
return false;
}
}
async disconnect(): Promise<void> {
if (this.keepaliveManager) {
await this.keepaliveManager.stop();
}
this.isConnected = false;
console.log('Disconnected from NextBlock');
}
getConnectionHealth(): ConnectionHealth | null {
return this.keepaliveManager?.getHealth() || null;
}
get connected(): boolean {
return this.isConnected;
}
// get grpcClient(): any {
// return this.client;
// }
}
// Health monitoring and alerting
class HealthMonitor {
private keepaliveManager: KeepaliveManager;
private alertThresholds = {
successRate: 90.0, // Alert if success rate < 90%
avgPingTime: 1000, // Alert if avg ping time > 1s
consecutiveFailures: 2, // Alert after 2 consecutive failures
};
private monitorInterval?: NodeJS.Timeout;
constructor(keepaliveManager: KeepaliveManager) {
this.keepaliveManager = keepaliveManager;
}
startMonitoring(checkInterval: number = 30000): void {
this.monitorInterval = setInterval(() => {
this.checkHealth();
}, checkInterval);
console.log(`Health monitoring started with ${checkInterval}ms interval`);
}
stopMonitoring(): void {
if (this.monitorInterval) {
clearInterval(this.monitorInterval);
this.monitorInterval = undefined;
}
console.log('Health monitoring stopped');
}
private checkHealth(): void {
const health = this.keepaliveManager.getHealth();
// Check success rate
const successRate = health.totalPingsSent > 0
? (health.totalPingsSuccessful / health.totalPingsSent) * 100
: 100;
if (successRate < this.alertThresholds.successRate) {
this.triggerAlert('Low success rate', `Success rate: ${successRate.toFixed(1)}%`);
}
// Check average ping time
if (health.averagePingTime > this.alertThresholds.avgPingTime) {
this.triggerAlert('High ping time', `Average ping time: ${health.averagePingTime.toFixed(1)}ms`);
}
// Check consecutive failures
if (health.consecutiveFailures >= this.alertThresholds.consecutiveFailures) {
this.triggerAlert('Connection issues', `Consecutive failures: ${health.consecutiveFailures}`);
}
}
private triggerAlert(alertType: string, details: string): void {
const timestamp = new Date().toISOString();
console.warn(`🚨 HEALTH ALERT [${timestamp}] ${alertType}: ${details}`);
// Implement additional alerting logic here
// - Send notifications
// - Post to monitoring systems
// - Trigger recovery procedures
}
}
## Usage Examples
// Basic keepalive example
async function basicKeepaliveExample(): Promise<void> {
console.log('Starting basic keepalive example...');
// Connect to NextBlock (see connection.md)
// const config = configFromEnv();
// const connection = await createNextBlockClient(config);
// Start keepalive task
const keepalivePromise = startKeepaliveTask(
// connection.client,
{
pingInterval: 30000, // Ping every 30 seconds
maxConsecutiveFailures: 3,
reconnectDelay: 5000,
healthCheckEnabled: true,
timeout: 15000,
}
);
console.log('Application running with keepalive...');
// Simulate application work
setTimeout(() => {
console.log('Application work completed, stopping...');
process.exit(0);
}, 300000); // Run for 5 minutes
await keepalivePromise;
}
// Advanced keepalive with health monitoring
async function advancedKeepaliveExample(): Promise<void> {
console.log('Starting advanced keepalive example...');
// Configuration
const keepaliveConfig: KeepaliveConfig = {
pingInterval: 30000, // Ping every 30 seconds
maxConsecutiveFailures: 3,
reconnectDelay: 10000, // Wait 10 seconds before reconnection
healthCheckEnabled: true,
timeout: 15000,
};
// Use connection manager with integrated keepalive
const manager = new ConnectionManagerWithKeepalive(
{}, // nextblockConfig placeholder
keepaliveConfig
);
try {
const connected = await manager.connect();
if (connected) {
console.log('Connected with keepalive enabled');
// Start health monitoring
const healthMonitor = new HealthMonitor(manager.keepaliveManager!);
healthMonitor.startMonitoring(60000); // Check every minute
// Simulate application work with periodic health checks
for (let i = 0; i < 10; i++) {
await new Promise(resolve => setTimeout(resolve, 30000));
const health = manager.getConnectionHealth();
if (health) {
console.log(`Connection health check ${i + 1}:`);
console.log(` Healthy: ${health.isHealthy}`);
console.log(` Success rate: ${((health.totalPingsSuccessful / Math.max(health.totalPingsSent, 1)) * 100).toFixed(1)}%`);
console.log(` Avg ping time: ${health.averagePingTime.toFixed(1)}ms`);
console.log(` Total pings: ${health.totalPingsSent}`);
}
}
healthMonitor.stopMonitoring();
} else {
console.error('Failed to establish connection');
}
} catch (error) {
console.error('Advanced keepalive example failed:', error);
} finally {
await manager.disconnect();
}
}
// Connection pool with keepalive
class ConnectionPoolWithKeepalive {
private connections: ConnectionManagerWithKeepalive[] = [];
private currentIndex: number = 0;
private poolSize: number;
private nextblockConfig: any;
private keepaliveConfig: KeepaliveConfig;
constructor(
nextblockConfig: any,
keepaliveConfig: KeepaliveConfig,
poolSize: number = 5
) {
this.nextblockConfig = nextblockConfig;
this.keepaliveConfig = keepaliveConfig;
this.poolSize = poolSize;
}
async initialize(): Promise<void> {
console.log(`Initializing connection pool with ${this.poolSize} connections...`);
const connectionPromises = Array.from({ length: this.poolSize }, async (_, i) => {
const manager = new ConnectionManagerWithKeepalive(
this.nextblockConfig,
this.keepaliveConfig
);
const connected = await manager.connect();
if (connected) {
this.connections.push(manager);
console.log(`Connection ${i + 1} established with keepalive`);
} else {
console.warn(`Failed to create connection ${i + 1}`);
}
});
await Promise.all(connectionPromises);
console.log(`Connection pool initialized with ${this.connections.length} connections`);
}
getConnection(): ConnectionManagerWithKeepalive {
if (this.connections.length === 0) {
throw new Error('No available connections in pool');
}
const connection = this.connections[this.currentIndex];
this.currentIndex = (this.currentIndex + 1) % this.connections.length;
return connection;
}
async closeAll(): Promise<void> {
await Promise.all(this.connections.map(conn => conn.disconnect()));
this.connections = [];
console.log('All connections closed');
}
getPoolHealth(): ConnectionHealth[] {
return this.connections
.map(conn => conn.getConnectionHealth())
.filter((health): health is ConnectionHealth => health !== null);
}
}
// Main example runner
async function main(): Promise<void> {
console.log('NextBlock Keepalive Examples');
const choice = process.argv[2] || '1';
switch (choice) {
case '1':
console.log('\n1. Basic Keepalive:');
await basicKeepaliveExample();
break;
case '2':
console.log('\n2. Advanced Keepalive with Monitoring:');
await advancedKeepaliveExample();
break;
case '3':
console.log('\n3. Connection Pool with Keepalive:');
const pool = new ConnectionPoolWithKeepalive(
{}, // config placeholder
defaultKeepaliveConfig,
3
);
await pool.initialize();
// Use the pool for some time
setTimeout(async () => {
const poolHealth = pool.getPoolHealth();
console.log(`Pool health summary: ${poolHealth.length} healthy connections`);
await pool.closeAll();
}, 60000);
break;
default:
console.log('Usage: node keepalive.js [1|2|3]');
console.log(' 1: Basic keepalive');
console.log(' 2: Advanced keepalive with monitoring');
console.log(' 3: Connection pool with keepalive');
}
}
if (require.main === module) {
main().catch(console.error);
}
export {
KeepaliveConfig,
ConnectionHealth,
KeepaliveManager,
ConnectionManagerWithKeepalive,
HealthMonitor,
ConnectionPoolWithKeepalive,
startKeepaliveTask,
};
## Best Practices
1. **Appropriate intervals**: Use 30-60 second ping intervals for most applications
2. **Health monitoring**: Track connection health and implement alerting
3. **Graceful recovery**: Handle connection failures with exponential backoff
4. **Resource cleanup**: Always stop keepalive tasks when shutting down
5. **Timeout handling**: Set reasonable timeouts for ping requests
6. **Logging**: Log keepalive events for debugging and monitoring
7. **Integration**: Integrate keepalive with your connection management system
8. **Connection pooling**: Use connection pools for high-throughput applications