From 5aa21260b98fb47c1b11b4685d45ffef8aed4f4e Mon Sep 17 00:00:00 2001 From: Ratan Kaliani Date: Thu, 21 Nov 2024 08:55:22 -0800 Subject: [PATCH 1/2] feat: overarching timeout --- Cargo.lock | 3 + script/Cargo.toml | 1 + script/bin/operator.rs | 130 +++++++++++++++++++++-------------------- script/src/util.rs | 2 +- 4 files changed, 73 insertions(+), 63 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 30cf771..da51e9c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1133,6 +1133,7 @@ dependencies = [ "tendermint", "tendermint-light-client-verifier", "tokio", + "tower 0.5.1", ] [[package]] @@ -6223,6 +6224,8 @@ version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2873938d487c3cfb9aed7546dc9f2711d867c9f90c46b889989a2cb84eba6b4f" dependencies = [ + "pin-project-lite", + "tokio", "tower-layer", "tower-service", ] diff --git a/script/Cargo.toml b/script/Cargo.toml index 9aafdbd..dcbd019 100644 --- a/script/Cargo.toml +++ b/script/Cargo.toml @@ -40,5 +40,6 @@ hex = "0.4.3" futures = "0.3.30" env_logger = "0.11.3" serde_json = "1" +tower = { version = "0.5.1", features = ["timeout"] } [build-dependencies] sp1-build = "3.0.0" diff --git a/script/bin/operator.rs b/script/bin/operator.rs index 3b7ed81..aabd4fb 100644 --- a/script/bin/operator.rs +++ b/script/bin/operator.rs @@ -221,70 +221,64 @@ impl SP1BlobstreamOperator { } } - async fn run(&mut self) -> Result<()> { - info!("Starting SP1 Blobstream operator"); - let mut fetcher = TendermintRPCClient::default(); - let loop_interval_mins = get_loop_interval_mins(); + async fn run(&self) -> Result<()> { + let fetcher = TendermintRPCClient::default(); let block_update_interval = get_block_update_interval(); - loop { - let contract = SP1Blobstream::new(self.contract_address, self.wallet_filler.clone()); - - // Read the data commitment max from the contract. - let data_commitment_max = contract - .DATA_COMMITMENT_MAX() - .call() - .await? - .DATA_COMMITMENT_MAX; - - // Get the latest block from the contract. - let current_block = contract.latestBlock().call().await?.latestBlock; - - // Get the head of the chain. - let latest_tendermint_block_nb = fetcher.get_latest_block_height().await; - - // Subtract 1 block to ensure the block is stable. - let latest_stable_tendermint_block = latest_tendermint_block_nb - 1; - - // block_to_request is the closest interval of block_interval less than min(latest_stable_tendermint_block, data_commitment_max + current_block) - let max_block = std::cmp::min( - latest_stable_tendermint_block, - data_commitment_max + current_block, - ); - let block_to_request = max_block - (max_block % block_update_interval); - - // If block_to_request is greater than the current block in the contract, attempt to request. - if block_to_request > current_block { - // The next block the operator should request. - let max_end_block = block_to_request; - - let target_block = fetcher - .find_block_to_request(current_block, max_end_block) - .await; - - info!("Current block: {}", current_block); - info!("Attempting to step to block {}", target_block); - - // Request a header range if the target block is not the next block. - match self.request_header_range(current_block, target_block).await { - Ok(proof) => { - let tx_hash = self.relay_header_range(proof).await?; - info!( - "Posted data commitment from block {} to block {}\nTransaction hash: {}", - current_block, target_block, tx_hash - ); - } - Err(e) => { - error!("Header range request failed: {}", e); - continue; - } - }; - } else { - info!("Next block to request is {} which is > the head of the Tendermint chain which is {}. Sleeping.", block_to_request + block_update_interval, latest_stable_tendermint_block); - } + let contract = SP1Blobstream::new(self.contract_address, self.wallet_filler.clone()); - tokio::time::sleep(tokio::time::Duration::from_secs(60 * loop_interval_mins)).await; + // Read the data commitment max from the contract. + let data_commitment_max = contract + .DATA_COMMITMENT_MAX() + .call() + .await? + .DATA_COMMITMENT_MAX; + + // Get the latest block from the contract. + let current_block = contract.latestBlock().call().await?.latestBlock; + + // Get the head of the chain. + let latest_tendermint_block_nb = fetcher.get_latest_block_height().await; + + // Subtract 1 block to ensure the block is stable. + let latest_stable_tendermint_block = latest_tendermint_block_nb - 1; + + // block_to_request is the closest interval of block_interval less than min(latest_stable_tendermint_block, data_commitment_max + current_block) + let max_block = std::cmp::min( + latest_stable_tendermint_block, + data_commitment_max + current_block, + ); + let block_to_request = max_block - (max_block % block_update_interval); + + // If block_to_request is greater than the current block in the contract, attempt to request. + if block_to_request > current_block { + // The next block the operator should request. + let max_end_block = block_to_request; + + let target_block = fetcher + .find_block_to_request(current_block, max_end_block) + .await; + + info!("Current block: {}", current_block); + info!("Attempting to step to block {}", target_block); + + // Request a header range if the target block is not the next block. + match self.request_header_range(current_block, target_block).await { + Ok(proof) => { + let tx_hash = self.relay_header_range(proof).await?; + info!( + "Posted data commitment from block {} to block {}\nTransaction hash: {}", + current_block, target_block, tx_hash + ); + } + Err(e) => { + return Err(anyhow::anyhow!("Header range request failed: {}", e)); + } + }; + } else { + info!("Next block to request is {} which is > the head of the Tendermint chain which is {}. Sleeping.", block_to_request + block_update_interval, latest_stable_tendermint_block); } + Ok(()) } } @@ -317,11 +311,23 @@ async fn main() { dotenv::dotenv().ok(); env_logger::init(); - let mut operator = SP1BlobstreamOperator::new().await; + let operator = SP1BlobstreamOperator::new().await; operator.check_vkey().await.unwrap(); + + info!("Starting SP1 Blobstream operator"); + const LOOP_TIMEOUT_MINS: u64 = 20; loop { - if let Err(e) = operator.run().await { + let request_interval_mins = get_loop_interval_mins(); + // If the operator takes longer than LOOP_TIMEOUT_MINS for a single invocation, or there's + // an error, sleep for the loop interval and try again. + if let Err(e) = tokio::time::timeout( + tokio::time::Duration::from_secs(60 * LOOP_TIMEOUT_MINS), + operator.run(), + ) + .await + { error!("Error running operator: {}", e); } + tokio::time::sleep(tokio::time::Duration::from_secs(60 * request_interval_mins)).await; } } diff --git a/script/src/util.rs b/script/src/util.rs index 7accfa2..ae1b61e 100644 --- a/script/src/util.rs +++ b/script/src/util.rs @@ -45,7 +45,7 @@ impl TendermintRPCClient { } // Search to find the greatest block number to request. - pub async fn find_block_to_request(&mut self, start_block: u64, max_end_block: u64) -> u64 { + pub async fn find_block_to_request(&self, start_block: u64, max_end_block: u64) -> u64 { let mut curr_end_block = max_end_block; loop { if curr_end_block - start_block == 1 { From bf190f7aaea549e18c9988f1eadc06955b3c2a50 Mon Sep 17 00:00:00 2001 From: Ratan Kaliani Date: Thu, 21 Nov 2024 08:55:48 -0800 Subject: [PATCH 2/2] fix --- Cargo.lock | 3 --- script/Cargo.toml | 1 - 2 files changed, 4 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index da51e9c..30cf771 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1133,7 +1133,6 @@ dependencies = [ "tendermint", "tendermint-light-client-verifier", "tokio", - "tower 0.5.1", ] [[package]] @@ -6224,8 +6223,6 @@ version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2873938d487c3cfb9aed7546dc9f2711d867c9f90c46b889989a2cb84eba6b4f" dependencies = [ - "pin-project-lite", - "tokio", "tower-layer", "tower-service", ] diff --git a/script/Cargo.toml b/script/Cargo.toml index dcbd019..9aafdbd 100644 --- a/script/Cargo.toml +++ b/script/Cargo.toml @@ -40,6 +40,5 @@ hex = "0.4.3" futures = "0.3.30" env_logger = "0.11.3" serde_json = "1" -tower = { version = "0.5.1", features = ["timeout"] } [build-dependencies] sp1-build = "3.0.0"