fix(pumpswap): harden quotes and low-latency examples

examples/pumpswap_trading/Cargo.toml

@@ -5,7 +5,7 @@ edition = "2021"
 
 [dependencies]
 sol-trade-sdk = { path = "../.." }
-solana-streamer-sdk = { version = "2.0.0", git = "https://github.com/0xfnzero/solana-streamer", rev = "85c6cc901ad3f1bf8fe8010d79b92ecdd0be02b4" }
+solana-streamer-sdk = { version = "2.0.0", git = "https://github.com/0xfnzero/solana-streamer", rev = "f1c6aecb3d4a4ebb2cd3c9f6a58de20b019418e2" }
 solana-sdk = "3.0.0"
 solana-commitment-config = { version = "3.0.0", features = ["serde"] }
 tokio = { version = "1", features = ["full"] }

examples/pumpswap_trading/README.md

@@ -22,7 +22,8 @@ cargo run --release --package pumpswap_trading
 - The buy uses `BuyAmount::WithMaxInput`, which applies slippage to maximum quote cost and is appropriate when fill priority matters.
 - Buy parameters use post-trade reserves and LP/protocol/creator fee bps from the event.
 - The event's raw and virtual quote reserves come from the same transaction snapshot. The hot path does not fetch the Pool account, avoiding both added latency and mixed-slot quotes.
-- The example records the pre-buy balance and sells only the confirmed balance increase. It refreshes pool state and blockhash before selling.
+- The first matching event asynchronously records the pre-buy balance; the next fresh event performs the trade without a balance RPC in the submission hot path. It sells only the confirmed balance increase and refreshes pool state and blockhash before selling.
 - Use `BuyAmount::ExactInput` when the quote spend must be exact. That mode protects minimum output and can fail more often in an active pool.
+- If baseline warmup fails, the example waits for another event. Once transaction execution starts, an error keeps the one-shot guard locked because submission or position state may be uncertain; inspect the signatures and account state before retrying.
 
 Production bots should also add durable signature deduplication, a position state machine, SWQoS configuration, and bounded requoting. Do not solve slippage errors by setting `min_out` to zero.

examples/pumpswap_trading/README_CN.md

@@ -20,7 +20,8 @@ cargo run --release --package pumpswap_trading
 - 买入使用 `BuyAmount::WithMaxInput`，适合优先成交的跟单/狙击场景，滑点限制最大 quote 成本。
 - 买入参数使用事件中的成交后储备和 LP/protocol/creator fee bps。
 - 原始 quote 储备和虚拟 quote 储备均来自同一笔交易的事件快照；热路径不再查询 Pool 账户，避免额外延迟和跨 slot 混合报价。
-- 示例记录买前余额，只卖出确认后的余额增量；卖出前重新获取池状态和 blockhash。
+- 第一个匹配事件用于异步记录买前余额，下一条新鲜事件才会交易，因此提交热路径不再查询余额；示例只卖出确认后的余额增量，并在卖出前重新获取池状态和 blockhash。
 - 若业务必须精确花费 quote，应改用 `BuyAmount::ExactInput`。这会启用最小输出保护，在活跃池中更容易因状态变化而失败。
+- 基线预热失败时会等待后续事件；一旦进入交易阶段，错误会保持单次执行锁定，因为提交状态或持仓可能不确定，必须先核对签名和账户状态再重试。
 
 生产机器人还应增加持久化签名去重、持仓状态机、SWQoS 配置和有限次数的重新报价。不要通过把 `min_out` 设为零来处理滑点错误。

examples/pumpswap_trading/src/main.rs

@@ -24,7 +24,7 @@ use solana_streamer_sdk::streaming::YellowstoneGrpc;
 use std::str::FromStr;
 use std::sync::{
     atomic::{AtomicBool, Ordering},
-    Arc,
+    Arc, RwLock,
 };
 use std::time::{Duration, Instant};
 use tokio::sync::watch;
@@ -91,6 +91,18 @@ struct CachedBlockhash {
     fetched_at: Instant,
 }
 
+#[derive(Clone, Copy)]
+struct PositionBaseline {
+    mint: Pubkey,
+    token_program: Pubkey,
+    amount: u64,
+}
+
+enum EventAction {
+    BaselineWarmed,
+    TradeCompleted,
+}
+
 #[derive(Clone)]
 struct BlockhashCache {
     receiver: watch::Receiver<CachedBlockhash>,
@@ -139,6 +151,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let trade_client = Arc::new(create_solana_trade_client().await?);
     let blockhash_cache = BlockhashCache::start(trade_client.infrastructure.rpc.clone()).await?;
+    let position_baseline = Arc::new(RwLock::new(None));
 
     let grpc = YellowstoneGrpc::new(
         std::env::var("GRPC_ENDPOINT")
@@ -146,7 +159,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         std::env::var("GRPC_AUTH_TOKEN").ok(),
     )?;
 
-    let callback = create_event_callback(trade_client, blockhash_cache, selection);
+    let callback =
+        create_event_callback(trade_client, blockhash_cache, position_baseline, selection);
     let protocols = vec![Protocol::PumpSwap];
     // Filter accounts
     let account_include = vec![
@@ -192,6 +206,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 fn create_event_callback(
     client: Arc<SolanaTrade>,
     blockhash_cache: BlockhashCache,
+    position_baseline: Arc<RwLock<Option<PositionBaseline>>>,
     selection: EventSelection,
 ) -> impl Fn(DexEvent) {
     move |event: DexEvent| match event {
@@ -207,15 +222,38 @@ fn create_event_callback(
                 return;
             }
             // Test code, only test one transaction
-            if !ALREADY_EXECUTED.swap(true, Ordering::SeqCst) {
+            if ALREADY_EXECUTED
+                .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
+                .is_ok()
+            {
                 let client = client.clone();
                 let blockhash_cache = blockhash_cache.clone();
+                let position_baseline = position_baseline.clone();
+                let was_preparing =
+                    position_baseline.read().map(|baseline| baseline.is_none()).unwrap_or(true);
                 tokio::spawn(async move {
-                    if let Err(err) =
-                        pumpswap_trade_with_grpc_buy_event(client, blockhash_cache, e).await
+                    match pumpswap_trade_with_grpc_buy_event(
+                        client,
+                        blockhash_cache,
+                        position_baseline,
+                        selection,
+                        e,
+                    )
+                    .await
                     {
-                        eprintln!("Error in trade: {:?}", err);
-                        std::process::exit(1);
+                        Ok(EventAction::BaselineWarmed) => {
+                            ALREADY_EXECUTED.store(false, Ordering::Release);
+                        }
+                        Ok(EventAction::TradeCompleted) => {}
+                        Err(err) if was_preparing => {
+                            eprintln!("baseline warmup failed; waiting for a later event: {err:?}");
+                            ALREADY_EXECUTED.store(false, Ordering::Release);
+                        }
+                        Err(err) => {
+                            eprintln!(
+                                "trade failed after entering execution state: {err:?}; automatic retry is disabled because submission status or position state may be uncertain"
+                            );
+                        }
                     }
                 });
             }
@@ -232,15 +270,38 @@ fn create_event_callback(
                 return;
             }
             // Test code, only test one transaction
-            if !ALREADY_EXECUTED.swap(true, Ordering::SeqCst) {
+            if ALREADY_EXECUTED
+                .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
+                .is_ok()
+            {
                 let client = client.clone();
                 let blockhash_cache = blockhash_cache.clone();
+                let position_baseline = position_baseline.clone();
+                let was_preparing =
+                    position_baseline.read().map(|baseline| baseline.is_none()).unwrap_or(true);
                 tokio::spawn(async move {
-                    if let Err(err) =
-                        pumpswap_trade_with_grpc_sell_event(client, blockhash_cache, e).await
+                    match pumpswap_trade_with_grpc_sell_event(
+                        client,
+                        blockhash_cache,
+                        position_baseline,
+                        selection,
+                        e,
+                    )
+                    .await
                     {
-                        eprintln!("Error in trade: {:?}", err);
-                        std::process::exit(1);
+                        Ok(EventAction::BaselineWarmed) => {
+                            ALREADY_EXECUTED.store(false, Ordering::Release);
+                        }
+                        Ok(EventAction::TradeCompleted) => {}
+                        Err(err) if was_preparing => {
+                            eprintln!("baseline warmup failed; waiting for a later event: {err:?}");
+                            ALREADY_EXECUTED.store(false, Ordering::Release);
+                        }
+                        Err(err) => {
+                            eprintln!(
+                                "trade failed after entering execution state: {err:?}; automatic retry is disabled because submission status or position state may be uncertain"
+                            );
+                        }
                     }
                 });
             }
@@ -274,8 +335,10 @@ async fn create_solana_trade_client() -> AnyResult<SolanaTrade> {
 async fn pumpswap_trade_with_grpc_buy_event(
     client: Arc<SolanaTrade>,
     blockhash_cache: BlockhashCache,
+    position_baseline: Arc<RwLock<Option<PositionBaseline>>>,
+    selection: EventSelection,
     trade_info: PumpSwapBuyEvent,
-) -> AnyResult<()> {
+) -> AnyResult<EventAction> {
     let params = PumpSwapParams::from_trade_with_fee_basis_points(
         trade_info.pool,
         trade_info.base_mint,
@@ -305,16 +368,25 @@ async fn pumpswap_trade_with_grpc_buy_event(
     } else {
         trade_info.base_mint
     };
-    pumpswap_trade_with_grpc(&client, &blockhash_cache, trade_info.metadata.recv_us, mint, params)
-        .await?;
-    Ok(())
+    pumpswap_trade_with_grpc(
+        &client,
+        &blockhash_cache,
+        &position_baseline,
+        trade_info.metadata.recv_us,
+        selection.max_event_age_ms,
+        mint,
+        params,
+    )
+    .await
 }
 
 async fn pumpswap_trade_with_grpc_sell_event(
     client: Arc<SolanaTrade>,
     blockhash_cache: BlockhashCache,
+    position_baseline: Arc<RwLock<Option<PositionBaseline>>>,
+    selection: EventSelection,
     trade_info: PumpSwapSellEvent,
-) -> AnyResult<()> {
+) -> AnyResult<EventAction> {
     let params = PumpSwapParams::from_trade_with_fee_basis_points(
         trade_info.pool,
         trade_info.base_mint,
@@ -344,24 +416,33 @@ async fn pumpswap_trade_with_grpc_sell_event(
     } else {
         trade_info.base_mint
     };
-    pumpswap_trade_with_grpc(&client, &blockhash_cache, trade_info.metadata.recv_us, mint, params)
-        .await?;
-    Ok(())
+    pumpswap_trade_with_grpc(
+        &client,
+        &blockhash_cache,
+        &position_baseline,
+        trade_info.metadata.recv_us,
+        selection.max_event_age_ms,
+        mint,
+        params,
+    )
+    .await
 }
 
 async fn pumpswap_trade_with_grpc(
     client: &SolanaTrade,
     blockhash_cache: &BlockhashCache,
+    position_baseline: &Arc<RwLock<Option<PositionBaseline>>>,
     grpc_recv_us: i64,
+    max_event_age_ms: u64,
     mint_pubkey: Pubkey,
     params: PumpSwapParams,
-) -> AnyResult<()> {
+) -> AnyResult<EventAction> {
     println!("Testing PumpSwap trading...");
+    validate_pumpswap_snapshot(&params)?;
+    if !is_event_fresh(grpc_recv_us, now_micros(), max_event_age_ms) {
+        anyhow::bail!("event became stale before transaction construction");
+    }
     let slippage_basis_points = Some(500);
-    let recent_blockhash = blockhash_cache.latest()?;
-
-    let gas_fee_strategy = sol_trade_sdk::common::GasFeeStrategy::new();
-    gas_fee_strategy.set_global_fee_strategy(150000, 150000, 500000, 500000, 0.001, 0.001);
 
     let is_sol = params.base_mint == sol_trade_sdk::constants::WSOL_TOKEN_ACCOUNT
         || params.quote_mint == sol_trade_sdk::constants::WSOL_TOKEN_ACCOUNT;
@@ -372,8 +453,35 @@ async fn pumpswap_trade_with_grpc(
     } else {
         anyhow::bail!("target mint {} does not belong to pool {}", mint_pubkey, params.pool);
     };
-    let balance_before =
-        client.get_payer_token_balance_with_program(&mint_pubkey, &program_id).await?;
+    let baseline = position_baseline
+        .read()
+        .map_err(|_| anyhow::anyhow!("position baseline lock is poisoned"))?
+        .as_ref()
+        .copied();
+    let balance_before = if let Some(baseline) = baseline {
+        if baseline.mint != mint_pubkey || baseline.token_program != program_id {
+            anyhow::bail!("cached position baseline belongs to a different mint or token program");
+        }
+        baseline.amount
+    } else {
+        let amount = client.get_payer_token_balance_with_program(&mint_pubkey, &program_id).await?;
+        let mut baseline = position_baseline
+            .write()
+            .map_err(|_| anyhow::anyhow!("position baseline lock is poisoned"))?;
+        *baseline = Some(PositionBaseline { mint: mint_pubkey, token_program: program_id, amount });
+        println!(
+            "Position baseline warmed at {} base units; waiting for the next fresh matching event",
+            amount
+        );
+        return Ok(EventAction::BaselineWarmed);
+    };
+
+    let recent_blockhash = blockhash_cache.latest()?;
+    let gas_fee_strategy = sol_trade_sdk::common::GasFeeStrategy::new();
+    gas_fee_strategy.set_global_fee_strategy(150000, 150000, 500000, 500000, 0.001, 0.001);
+    if !is_event_fresh(grpc_recv_us, now_micros(), max_event_age_ms) {
+        anyhow::bail!("event became stale while preparing the transaction");
+    }
 
     // Buy tokens
     println!("Buying tokens from PumpSwap...");
@@ -431,8 +539,32 @@ async fn pumpswap_trade_with_grpc(
         anyhow::bail!("sell failed: {:?}; signatures: {:?}", err, sigs);
     }
 
-    // Exit program
-    std::process::exit(0);
+    println!("Round-trip example completed; further matching events remain locked out");
+    Ok(EventAction::TradeCompleted)
+}
+
+fn validate_pumpswap_snapshot(params: &PumpSwapParams) -> AnyResult<()> {
+    let required = [
+        ("pool", params.pool),
+        ("base_mint", params.base_mint),
+        ("quote_mint", params.quote_mint),
+        ("pool_base_token_account", params.pool_base_token_account),
+        ("pool_quote_token_account", params.pool_quote_token_account),
+        ("coin_creator_vault_ata", params.coin_creator_vault_ata),
+        ("coin_creator_vault_authority", params.coin_creator_vault_authority),
+        ("base_token_program", params.base_token_program),
+        ("quote_token_program", params.quote_token_program),
+    ];
+    for (name, value) in required {
+        if value == Pubkey::default() {
+            anyhow::bail!("event snapshot is missing {name}");
+        }
+    }
+    if params.pool_base_token_reserves == 0 || params.pool_quote_token_reserves == 0 {
+        anyhow::bail!("event snapshot has an empty raw pool reserve");
+    }
+    params.effective_quote_reserves()?;
+    Ok(())
 }
 
 #[cfg(test)]