feat: make priority fees optional and unify send transaction logic between v1, v2, and compression

.github/workflows/forester-tests.yml

@@ -45,7 +45,7 @@ jobs:
   test:
     name: Forester e2e test
     runs-on: ubuntu-latest
-    timeout-minutes: 60
+    timeout-minutes: 120
 
     services:
       redis:

cli/src/utils/constants.ts

@@ -21,8 +21,8 @@ export const LIGHT_PROVER_PROCESS_NAME = "light-prover";
 export const INDEXER_PROCESS_NAME = "photon";
 export const FORESTER_PROCESS_NAME = "forester";
 
-export const SURFPOOL_VERSION = "1.0.1";
-export const SURFPOOL_RELEASE_TAG = "v1.0.1-light";
+export const SURFPOOL_VERSION = "1.1.1";
+export const SURFPOOL_RELEASE_TAG = "v1.1.1-light";
 
 // PHOTON_VERSION, PHOTON_GIT_COMMIT, and PHOTON_GIT_REPO are auto-generated
 // from the external/photon submodule at build time. See photonVersion.generated.ts.

forester/Cargo.toml

@@ -35,7 +35,6 @@ light-token = { workspace = true }
 light-compressed-token-sdk = { workspace = true }
 solana-rpc-client-api = { workspace = true }
 solana-transaction-status = { workspace = true }
-bb8 = { workspace = true }
 base64 = { workspace = true }
 serde_json = { workspace = true }
 serde = { workspace = true }

forester/README.md

@@ -100,6 +100,7 @@ When a fallback RPC URL is configured, the pool automatically switches to it if
 | `--legacy-ixs-per-tx` | `LEGACY_IXS_PER_TX` | 1 | Instructions per V1 transaction |
 | `--cu-limit` | `CU_LIMIT` | 1000000 | Compute unit limit per transaction |
 | `--enable-priority-fees` | `ENABLE_PRIORITY_FEES` | false | Enable dynamic priority fees |
+| `--priority-fee-microlamports` | `PRIORITY_FEE_MICROLAMPORTS` | | Fixed priority fee in micro-lamports per compute unit |
 | `--lookup-table-address` | `LOOKUP_TABLE_ADDRESS` | | Address lookup table for versioned transactions |
 | `--helius-rpc` | `HELIUS_RPC` | false | Use Helius `getProgramAccountsV2` |
 
@@ -144,7 +145,7 @@ cargo run -- start \
   --processor-mode v2 \
   --max-concurrent-sends 200 \
   --cu-limit 400000 \
-  --enable-priority-fees true
+  --priority-fee-microlamports 10000
 ```
 
 ## Status & Health

forester/src/api_server.rs

@@ -1,4 +1,10 @@
-use std::{collections::HashMap, net::SocketAddr, sync::Arc, thread::JoinHandle, time::Duration};
+use std::{
+    collections::HashMap,
+    net::SocketAddr,
+    sync::{mpsc, Arc},
+    thread::JoinHandle,
+    time::Duration,
+};
 
 use serde::{Deserialize, Serialize};
 use tokio::sync::{broadcast, oneshot, watch};
@@ -1025,21 +1031,25 @@ async fn run_compressible_provider(
 ///
 /// # Returns
 /// An `ApiServerHandle` that can be used to trigger graceful shutdown
-pub fn spawn_api_server(config: ApiServerConfig) -> ApiServerHandle {
+pub fn spawn_api_server(config: ApiServerConfig) -> anyhow::Result<ApiServerHandle> {
     let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
+    let (startup_tx, startup_rx) = mpsc::sync_channel(1);
     let run_id_for_handle = config.run_id.clone();
 
     let thread_handle = std::thread::spawn(move || {
         let run_id = config.run_id.clone();
         let rt = match tokio::runtime::Runtime::new() {
             Ok(rt) => rt,
-            Err(e) => {
+            Err(error) => {
                 error!(
                     event = "api_server_runtime_create_failed",
                     run_id = %run_id,
-                    error = %e,
+                    error = %error,
                     "Failed to create tokio runtime for API server"
                 );
+                let _ = startup_tx.send(Err(anyhow::anyhow!(
+                    "Failed to create tokio runtime for API server: {error}"
+                )));
                 return;
             }
         };
@@ -1063,10 +1073,24 @@ pub fn spawn_api_server(config: ApiServerConfig) -> ApiServerHandle {
             );
 
             // Shared HTTP client with timeout for external requests (Prometheus)
-            let http_client = reqwest::Client::builder()
+            let http_client = match reqwest::Client::builder()
                 .timeout(EXTERNAL_HTTP_TIMEOUT)
                 .build()
-                .expect("Failed to create HTTP client");
+            {
+                Ok(client) => client,
+                Err(error) => {
+                    error!(
+                        event = "api_server_http_client_create_failed",
+                        run_id = %run_id,
+                        error = %error,
+                        "Failed to create shared HTTP client for API server"
+                    );
+                    let _ = startup_tx.send(Err(anyhow::anyhow!(
+                        "Failed to create shared HTTP client for API server: {error}"
+                    )));
+                    return;
+                }
+            };
 
             // Build trackers from config
             let trackers = config
@@ -1193,9 +1217,27 @@ pub fn spawn_api_server(config: ApiServerConfig) -> ApiServerHandle {
                 .or(compressible_route)
                 .with(cors);
 
+            let listener = match tokio::net::TcpListener::bind(addr).await {
+                Ok(listener) => listener,
+                Err(error) => {
+                    error!(
+                        event = "api_server_bind_failed",
+                        run_id = %run_id,
+                        address = %addr,
+                        error = %error,
+                        "Failed to bind API server socket"
+                    );
+                    let _ = startup_tx.send(Err(anyhow::anyhow!(
+                        "Failed to bind API server to {addr}: {error}"
+                    )));
+                    return;
+                }
+            };
+
+            let _ = startup_tx.send(Ok(()));
+
             warp::serve(routes)
-                .bind(addr)
-                .await
+                .incoming(listener)
                 .graceful({
                     let run_id_for_shutdown = run_id.clone();
                     async move {
@@ -1219,10 +1261,22 @@ pub fn spawn_api_server(config: ApiServerConfig) -> ApiServerHandle {
         });
     });
 
-    ApiServerHandle {
-        thread_handle,
-        shutdown_tx,
-        run_id: run_id_for_handle,
+    match startup_rx.recv() {
+        Ok(Ok(())) => Ok(ApiServerHandle {
+            thread_handle,
+            shutdown_tx,
+            run_id: run_id_for_handle,
+        }),
+        Ok(Err(error)) => {
+            let _ = thread_handle.join();
+            Err(error)
+        }
+        Err(_) => {
+            let _ = thread_handle.join();
+            Err(anyhow::anyhow!(
+                "API server startup acknowledgment channel closed before initialization completed"
+            ))
+        }
     }
 }
 

forester/src/cli.rs

@@ -150,9 +150,24 @@ pub struct StartArgs {
     #[arg(long, env = "CU_LIMIT", default_value = "1000000")]
     pub cu_limit: u32,
 
-    #[arg(long, env = "ENABLE_PRIORITY_FEES", default_value = "false")]
+    #[arg(
+        long,
+        env = "ENABLE_PRIORITY_FEES",
+        default_value = "false",
+        conflicts_with = "priority_fee_microlamports",
+        help = "Enable dynamic priority fees via RPC estimation"
+    )]
     pub enable_priority_fees: bool,
 
+    #[arg(
+        long,
+        env = "PRIORITY_FEE_MICROLAMPORTS",
+        value_parser = clap::value_parser!(u64).range(1..),
+        conflicts_with = "enable_priority_fees",
+        help = "Fixed priority fee in micro-lamports per compute unit"
+    )]
+    pub priority_fee_microlamports: Option<u64>,
+
     #[arg(long, env = "RPC_POOL_SIZE", default_value = "100")]
     pub rpc_pool_size: u32,
 
@@ -525,6 +540,34 @@ mod tests {
         assert!(result.is_err());
     }
 
+    #[test]
+    fn test_priority_fee_parsing() {
+        let args = StartArgs::try_parse_from([
+            "forester",
+            "--rpc-url", "http://test.com",
+            "--indexer-url", "http://indexer.test.com",
+            "--payer", "[1,2,3]",
+            "--derivation", "[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32]",
+            "--priority-fee-microlamports", "25000"
+        ]).unwrap();
+        assert_eq!(args.priority_fee_microlamports, Some(25_000));
+        assert!(!args.enable_priority_fees);
+    }
+
+    #[test]
+    fn test_priority_fee_modes_conflict() {
+        let result = StartArgs::try_parse_from([
+            "forester",
+            "--rpc-url", "http://test.com",
+            "--indexer-url", "http://indexer.test.com",
+            "--payer", "[1,2,3]",
+            "--derivation", "[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32]",
+            "--enable-priority-fees", "true",
+            "--priority-fee-microlamports", "25000"
+        ]);
+        assert!(result.is_err());
+    }
+
     #[test]
     fn test_processor_mode_display() {
         assert_eq!(ProcessorMode::V1.to_string(), "v1");

forester/src/compressible/ctoken/compressor.rs

@@ -22,6 +22,7 @@ use tracing::debug;
 use super::{state::CTokenAccountTracker, types::CTokenAccountState};
 use crate::{
     compressible::traits::{send_and_confirm_with_tracking, CompressibleTracker},
+    smart_transaction::TransactionPolicy,
     Result,
 };
 
@@ -30,6 +31,7 @@ pub struct CTokenCompressor<R: Rpc + Indexer> {
     rpc_pool: Arc<SolanaRpcPool<R>>,
     tracker: Arc<CTokenAccountTracker>,
     payer_keypair: Keypair,
+    transaction_policy: TransactionPolicy,
 }
 
 impl<R: Rpc + Indexer> Clone for CTokenCompressor<R> {
@@ -38,6 +40,7 @@ impl<R: Rpc + Indexer> Clone for CTokenCompressor<R> {
             rpc_pool: Arc::clone(&self.rpc_pool),
             tracker: Arc::clone(&self.tracker),
             payer_keypair: self.payer_keypair.insecure_clone(),
+            transaction_policy: self.transaction_policy,
         }
     }
 }
@@ -47,11 +50,13 @@ impl<R: Rpc + Indexer> CTokenCompressor<R> {
         rpc_pool: Arc<SolanaRpcPool<R>>,
         tracker: Arc<CTokenAccountTracker>,
         payer_keypair: Keypair,
+        transaction_policy: TransactionPolicy,
     ) -> Self {
         Self {
             rpc_pool,
             tracker,
             payer_keypair,
+            transaction_policy,
         }
     }
 
@@ -249,6 +254,7 @@ impl<R: Rpc + Indexer> CTokenCompressor<R> {
             &mut *rpc,
             &[ix],
             &self.payer_keypair,
+            self.transaction_policy,
             &*self.tracker,
             &pubkeys,
             "compress_and_close",