feat: Update full test to use tycho-test

- It uses the index command of tycho indexer to sync the protocol until now and then stream the live updates
- Connects to the stream and processes updates
- Calculates spot prices, limits and get amount out using Tycho Simulation
- It simulates execution using RPC batched RPC requests
  - Uses router and executor overwrites (so that the executor doesn't need to be deployed)
  - It batches the simulations in smaller chunks. The requests are really big because of the overwrites

Other changes:
- Using ProtocolStreamBuilder in the state registry instead of TychoStreamDecoder. Then in the ranged test we are just extracting the decoder from the ProtocolStreamBuilder (hacky but it works)
- In the full test we need to keep track of the ProtocolComponents to simulate in multiple blocks
- Changed get_current_block to get_block(block_number)

#time 7h 5m

#time 8m

Author: dianacarvalho1

protocol-testing/Cargo.lock

@@ -8,23 +8,25 @@ glob = "0.3.0"
 miette = { version = "7.6.0", features = ["fancy"] }
 # Logging & Tracing
 tracing = "0.1.37"
-tycho-simulation = { git = "https://github.com/propeller-heads/tycho-simulation.git", tag = "0.196.4", features = ["evm"] }
-tycho-test = { git = "https://github.com/propeller-heads/tycho-simulation.git", tag = "0.196.4" }
-tycho-execution = ">=0.134.0"
+tycho-simulation = { path = "../../tycho-simulation", features = ["evm"] }
+tycho-test = { path = "../../tycho-simulation/tycho-test" }
+tycho-execution = { path = "../../tycho-execution" }
 num-bigint = "0.4"
 num-traits = "0.2"
 num-rational = "0.4.2"
 # EVM dependencies
 alloy = { version = "1.0.30", features = ["arbitrary", "json", "dyn-abi", "sol-types", "contract", "provider-http", "providers", "signer-local", "rpc-types-eth", "rpc-types-trace"] }
 reqwest = { version = "0.12", features = ["json"] }
 tokio = { version = "1", features = ["full"] }
+futures = "0.3"
 serde_json = "1.0.140"
 clap = "4.5.31"
 figment = { version = "0.10.19", features = ["yaml"] }
 serde = { version = "1.0.218", features = ["derive"] }
 hex = "0.4.3"
 tracing-subscriber = "0.3.19"
 postgres = "0.19.10"
+tokio-postgres = "0.7"
 serde_yaml = "0.9.34"
 dotenv = "0.15.0"
 async-trait = "0.1.87"
@@ -33,4 +35,5 @@ similar = "2.7.0"
 termsize = "0.1.9"
 itertools = "0.14.0"
 tokio-retry2 = { version = "0.6.0", features = ["jitter"] }
-regex = "1.11.1"
+regex = "1.11.1"
+tempfile = "3.8.1"

protocol-testing/Cargo.toml

@@ -31,7 +31,7 @@ enum TestSubcommand {
     Range(RangeTestCommand),
 }
 
-/// Run a test from a specific initial block to the latest block
+/// Run continuous sync test from a specific initial block
 #[derive(Args)]
 pub struct FullTestCommand {
     #[command(flatten)]
@@ -42,22 +42,13 @@ pub struct FullTestCommand {
     /// configuration.
     #[arg(long)]
     initial_block: Option<u64>,
-
-    /// Stop the test at this block number.
-    /// If not provided, it will use the latest block of the chain at the time of running the
-    /// command.
-    #[arg(long)]
-    stop_block: Option<u64>,
 }
 
 impl FullTestCommand {
     fn run(self) -> miette::Result<()> {
         let args = self.common_args;
         TestRunner::new(
-            TestType::Full(TestTypeFull {
-                initial_block: self.initial_block,
-                stop_block: self.stop_block,
-            }),
+            TestType::Full(TestTypeFull { initial_block: self.initial_block }),
             args.root_path()?,
             args.chain,
             args.package,

protocol-testing/src/main.rs

@@ -83,14 +83,14 @@ impl RPCProvider {
             .and_then(|block_opt| block_opt.ok_or_else(|| miette::miette!("Block not found")))
     }
 
-    pub async fn get_current_block(&self) -> miette::Result<Block> {
-        info!("Fetching current block...");
+    pub async fn get_block(&self, block_number: BlockNumberOrTag) -> miette::Result<Block> {
+        info!("Fetching block {:?}...", block_number.as_number());
         let provider = ProviderBuilder::new().connect_http(self.url.clone());
         provider
-            .get_block_by_number(BlockNumberOrTag::Latest)
+            .get_block_by_number(block_number)
             .await
             .into_diagnostic()
-            .wrap_err("Failed to fetch current block")
+            .wrap_err(format!("Failed to fetch block {:?}", block_number))
             .and_then(|block_opt| block_opt.ok_or_else(|| miette::miette!("Block not found")))
     }
 }

protocol-testing/src/rpc.rs

@@ -1,50 +1,60 @@
 use tycho_simulation::{
     evm::{
-        decoder::TychoStreamDecoder,
         engine_db::tycho_db::PreCachedDB,
         protocol::{
             ekubo::state::EkuboState, pancakeswap_v2::state::PancakeswapV2State,
             uniswap_v2::state::UniswapV2State, uniswap_v3::state::UniswapV3State,
             vm::state::EVMPoolState,
         },
+        stream::ProtocolStreamBuilder,
     },
     protocol::models::DecoderContext,
-    tycho_client::feed::BlockHeader,
+    tycho_client::feed::component_tracker::ComponentFilter,
 };
 
 /// Register decoder based on protocol system. Defaults to EVMPoolState.
 /// To add a new protocol, just add a case to the match statement.
-pub fn register_decoder_for_protocol(
-    decoder: &mut TychoStreamDecoder<BlockHeader>,
+pub fn register_protocol(
+    stream_builder: ProtocolStreamBuilder,
     protocol_system: &str,
     decoder_context: DecoderContext,
-) -> miette::Result<()> {
-    match protocol_system {
-        "uniswap_v2" | "sushiswap_v2" => {
-            decoder
-                .register_decoder_with_context::<UniswapV2State>(protocol_system, decoder_context);
-        }
-        "pancakeswap_v2" => {
-            decoder.register_decoder_with_context::<PancakeswapV2State>(
+) -> miette::Result<ProtocolStreamBuilder> {
+    let tvl_filter = ComponentFilter::with_tvl_range(100.0, 100.0);
+    let stream_builder = match protocol_system {
+        "uniswap_v2" | "sushiswap_v2" => stream_builder
+            .exchange_with_decoder_context::<UniswapV2State>(
                 protocol_system,
+                tvl_filter,
+                None,
                 decoder_context,
-            );
-        }
-        "uniswap_v3" | "pancakeswap_v3" => {
-            decoder
-                .register_decoder_with_context::<UniswapV3State>(protocol_system, decoder_context);
-        }
-        "ekubo_v2" => {
-            decoder.register_decoder_with_context::<EkuboState>(protocol_system, decoder_context);
-        }
-        // Default to EVMPoolState for all other protocols
-        _ => {
-            decoder.register_decoder_with_context::<EVMPoolState<PreCachedDB>>(
+            ),
+        "pancakeswap_v2" => stream_builder.exchange_with_decoder_context::<PancakeswapV2State>(
+            protocol_system,
+            tvl_filter,
+            None,
+            decoder_context,
+        ),
+        "uniswap_v3" | "pancakeswap_v3" => stream_builder
+            .exchange_with_decoder_context::<UniswapV3State>(
                 protocol_system,
+                tvl_filter,
+                None,
                 decoder_context,
-            );
-        }
-    }
+            ),
+        "ekubo_v2" => stream_builder.exchange_with_decoder_context::<EkuboState>(
+            protocol_system,
+            tvl_filter,
+            None,
+            decoder_context,
+        ),
+        // Default to EVMPoolState for all other protocols
+        _ => stream_builder.exchange_with_decoder_context::<EVMPoolState<PreCachedDB>>(
+            protocol_system,
+            tvl_filter,
+            None,
+            decoder_context,
+        ),
+    };
 
-    Ok(())
+    Ok(stream_builder)
 }