Merge remote-tracking branch 'origin/main-v0.14.2' into shahak/merge-main-v0.14.1-committer-into-main-v0.14.2-1771144428

Author: ArniStarkware

Cargo.lock

@@ -3,6 +3,12 @@ use serde::{Serialize, Serializer};
 const IDLE_CONNECTIONS_FOR_AUTO_SCALED_SERVICES: usize = 0;
 const IDLE_CONNECTIONS_FOR_STATICALLY_SCALED_SERVICES: usize = 10;
 
+// Note: we explicitly use a new connection when sending a request to an autoscaled or a
+// service allowed to run on a spot instance to ensure each new request goes through the load
+// balancer and directed at an available server. This allows us to avoid having to address
+// connection termination issues, e.g., when the server is marked for eviction, and should be in
+// a graceful shutdown flow, and as such should reject new requests.
+
 /// Whether a service is autoscaled or not.
 #[derive(Clone, Debug, PartialEq)]
 pub enum ScalePolicy {

crates/apollo_deployments/src/scale_policy.rs

@@ -3,7 +3,6 @@ use std::fmt::Display;
 use std::fs::File;
 use std::io::Read;
 use std::iter::once;
-use std::net::{IpAddr, Ipv4Addr};
 use std::path::{Path, PathBuf};
 
 use apollo_config::dumping::{prepend_sub_config_name, ser_param, SerializeConfig};
@@ -377,7 +376,6 @@ pub(crate) trait GetComponentConfigs: ServiceNameInner {
     fn component_config_for_local_service(&self, port: u16) -> ReactiveComponentExecutionConfig {
         ReactiveComponentExecutionConfig::local_with_remote_enabled(
             REMOTE_SERVICE_URL_PLACEHOLDER.to_string(),
-            IpAddr::from(Ipv4Addr::UNSPECIFIED),
             port,
         )
     }
@@ -386,13 +384,9 @@ pub(crate) trait GetComponentConfigs: ServiceNameInner {
     fn component_config_for_remote_service(&self, port: u16) -> ReactiveComponentExecutionConfig {
         let idle_connections = self.get_scale_policy().idle_connections();
         let retries = self.get_retries();
-        ReactiveComponentExecutionConfig::remote(
-            REMOTE_SERVICE_URL_PLACEHOLDER.to_string(),
-            IpAddr::from(Ipv4Addr::UNSPECIFIED),
-            port,
-        )
-        .with_idle_connections(idle_connections)
-        .with_retries(retries)
+        ReactiveComponentExecutionConfig::remote(REMOTE_SERVICE_URL_PLACEHOLDER.to_string(), port)
+            .with_idle_connections(idle_connections)
+            .with_retries(retries)
     }
 
     fn component_config_pair(&self, port: u16) -> ComponentConfigPair {

crates/apollo_deployments/src/service.rs

@@ -1,5 +1,5 @@
 use std::collections::BTreeMap;
-use std::net::{IpAddr, ToSocketAddrs};
+use std::net::ToSocketAddrs;
 
 use apollo_config::dumping::{ser_optional_sub_config, ser_param, SerializeConfig};
 use apollo_config::{ParamPath, ParamPrivacyInput, SerializedParam};
@@ -135,8 +135,7 @@ impl ReactiveComponentExecutionConfig {
         }
     }
 
-    // TODO(Tsabary): remove the unused `_ip` arg.
-    pub fn remote(url: String, _ip: IpAddr, port: u16) -> Self {
+    pub fn remote(url: String, port: u16) -> Self {
         Self {
             execution_mode: ReactiveComponentExecutionMode::Remote,
             local_server_config: None,
@@ -148,8 +147,7 @@ impl ReactiveComponentExecutionConfig {
         }
     }
 
-    // TODO(Tsabary): remove the unused `_ip` arg.
-    pub fn local_with_remote_enabled(url: String, _ip: IpAddr, port: u16) -> Self {
+    pub fn local_with_remote_enabled(url: String, port: u16) -> Self {
         Self {
             execution_mode: ReactiveComponentExecutionMode::LocalExecutionWithRemoteEnabled,
             local_server_config: Some(LocalServerConfig::default()),

crates/apollo_node_config/src/component_execution_config.rs

@@ -220,7 +220,6 @@ pub async fn run_server(
 ) -> anyhow::Result<(SocketAddr, ServerHandle)> {
     debug!("Started get_last_synced_block");
     let starting_block = get_last_synced_block(storage_reader.clone())?;
-    debug!("Starting JSON-RPC.");
     let methods = get_methods_from_supported_apis(
         &config.chain_id,
         config.execution_config,
@@ -246,6 +245,7 @@ pub async fn run_server(
         .set_http_middleware(ServiceBuilder::new().filter_async(proxy_rpc_request));
 
     let server_address = SocketAddr::new(config.ip, config.port);
+    debug!("Starting JSON-RPC at address {server_address}.");
 
     if config.collect_metrics {
         let server = server_builder

crates/apollo_rpc/src/lib.rs

@@ -130,8 +130,6 @@ pub trait StorageReaderServerHandler<Request, Response> {
     ) -> Result<Response, StorageError>;
 }
 
-// TODO(Nadin): Remove #[allow(dead_code)] once the fields are used in the implementation.
-#[allow(dead_code)]
 /// A server for handling remote storage reader queries via a configurable request handler.
 pub struct StorageReaderServer<RequestHandler, Request, Response>
 where

crates/apollo_storage/src/storage_reader_server.rs

@@ -16,6 +16,7 @@ use starknet_api::core::ChainId;
 use starknet_api::transaction::Transaction;
 use state_reader::config::RpcStateReaderConfig;
 use state_reader::rpc_state_reader::ConsecutiveRpcStateReaders;
+use utils::get_chain_info;
 
 /// Executes a single transaction at the given block number using the RPC state reader.
 pub fn execute_single_transaction(
@@ -38,10 +39,11 @@ pub fn execute_single_transaction(
     assert!(block_number.0 != 0, "Cannot execute transaction at block 0");
     let prev_block_number = BlockNumber(block_number.0 - 1);
 
+    let chain_info = get_chain_info(&chain_id, None);
     let readers = ConsecutiveRpcStateReaders::new(
         prev_block_number,
         Some(rpc_state_reader_config),
-        chain_id,
+        chain_info,
         false,
         contract_class_manager,
     );

crates/blockifier_reexecution/src/lib.rs

@@ -14,6 +14,7 @@ use blockifier_reexecution::state_reader::config::RpcStateReaderConfig;
 use blockifier_reexecution::state_reader::offline_state_reader::OfflineConsecutiveStateReaders;
 use blockifier_reexecution::state_reader::reexecution_state_reader::ConsecutiveReexecutionStateReaders;
 use blockifier_reexecution::state_reader::rpc_state_reader::ConsecutiveRpcStateReaders;
+use blockifier_reexecution::utils::get_chain_info;
 use clap::Parser;
 use google_cloud_storage::client::{Client, ClientConfig};
 use google_cloud_storage::http::objects::download::Range;
@@ -68,10 +69,11 @@ async fn main() {
             // for details), so should be executed in a blocking thread.
             // TODO(Aner): make only the RPC calls blocking, not the whole function.
             tokio::task::spawn_blocking(move || {
+                let chain_info = get_chain_info(&rpc_args.parse_chain_id(), None);
                 ConsecutiveRpcStateReaders::new(
                     BlockNumber(block_number - 1),
                     Some(rpc_state_reader_config),
-                    rpc_args.parse_chain_id(),
+                    chain_info,
                     false,
                     contract_class_manager,
                 )
@@ -98,7 +100,7 @@ async fn main() {
             );
 
             let rpc_state_reader_config = RpcStateReaderConfig::from_url(rpc_args.node_url.clone());
-            let chain_id = rpc_args.parse_chain_id();
+            let chain_info = get_chain_info(&rpc_args.parse_chain_id(), None);
 
             // RPC calls are "synchronous IO" (see, e.g., https://stackoverflow.com/questions/74547541/when-should-you-use-tokios-spawn_blocking)
             // for details), so should be executed in a blocking thread.
@@ -107,7 +109,7 @@ async fn main() {
                 ConsecutiveRpcStateReaders::new(
                     BlockNumber(block_number - 1),
                     Some(rpc_state_reader_config),
-                    chain_id,
+                    chain_info,
                     false,
                     contract_class_manager,
                 )
@@ -129,7 +131,8 @@ async fn main() {
             let mut task_set = tokio::task::JoinSet::new();
             for block_number in block_numbers {
                 let full_file_path = block_full_file_path(directory_path.clone(), block_number);
-                let (node_url, chain_id) = (rpc_args.node_url.clone(), rpc_args.parse_chain_id());
+                let node_url = rpc_args.node_url.clone();
+                let chain_info = get_chain_info(&rpc_args.parse_chain_id(), None);
                 let contract_class_manager = contract_class_manager.clone();
 
                 // RPC calls are "synchronous IO" (see, e.g., https://stackoverflow.com/questions/74547541/when-should-you-use-tokios-spawn-blocking)
@@ -142,7 +145,7 @@ async fn main() {
                         ConsecutiveRpcStateReaders::new(
                             block_number.prev().expect("Should not run with block 0"),
                             Some(rpc_state_reader_config),
-                            chain_id,
+                            chain_info,
                             true,
                             contract_class_manager,
                         )

crates/blockifier_reexecution/src/main.rs

@@ -11,7 +11,7 @@ use blockifier::blockifier::transaction_executor::{
 };
 use blockifier::blockifier_versioned_constants::VersionedConstants;
 use blockifier::bouncer::BouncerConfig;
-use blockifier::context::BlockContext;
+use blockifier::context::{BlockContext, ChainInfo};
 use blockifier::execution::contract_class::RunnableCompiledClass;
 use blockifier::state::cached_state::{CommitmentStateDiff, StateMaps};
 use blockifier::state::contract_class_manager::ContractClassManager;
@@ -119,9 +119,8 @@ pub struct RpcStateReader {
     pub config: RpcStateReaderConfig,
     pub block_id: BlockId,
     pub(crate) retry_config: RetryConfig,
-    pub chain_id: ChainId,
-    /// Optional override for the STRK fee token address.
-    pub strk_fee_token_address: Option<ContractAddress>,
+
+    pub chain_info: ChainInfo,
     #[allow(dead_code)]
     pub(crate) contract_class_mapping_dumper: Arc<Mutex<Option<StarknetContractClassMapping>>>,
 }
@@ -133,8 +132,7 @@ impl Default for RpcStateReader {
             config,
             block_id: BlockId::Latest,
             retry_config: RetryConfig::default(),
-            chain_id: ChainId::Mainnet,
-            strk_fee_token_address: None,
+            chain_info: get_chain_info(&ChainId::Mainnet, None),
             contract_class_mapping_dumper: Arc::new(Mutex::new(None)),
         }
     }
@@ -143,7 +141,7 @@ impl Default for RpcStateReader {
 impl RpcStateReader {
     pub fn new(
         config: &RpcStateReaderConfig,
-        chain_id: ChainId,
+        chain_info: ChainInfo,
         block_number: BlockNumber,
         dump_mode: bool,
     ) -> Self {
@@ -155,34 +153,35 @@ impl RpcStateReader {
             config: config.clone(),
             block_id: BlockId::Number(block_number),
             retry_config: RetryConfig::default(),
-            chain_id,
-            strk_fee_token_address: None,
+            chain_info,
             contract_class_mapping_dumper,
         }
     }
 
-    /// Creates an RpcStateReader from a node URL, chain ID, and block ID.
-    /// Optionally accepts a STRK fee token address override for custom environments.
+    /// Creates an RpcStateReader from a node URL, chain info, and block ID.
     pub fn new_with_config_from_url(
         node_url: String,
-        chain_id: ChainId,
+        chain_info: ChainInfo,
         block_id: BlockId,
-        strk_fee_token_address: Option<ContractAddress>,
     ) -> Self {
         let config = RpcStateReaderConfig::from_url(node_url);
         let contract_class_mapping_dumper = Arc::new(Mutex::new(None));
         Self {
             config,
             block_id,
             retry_config: RetryConfig::default(),
-            chain_id,
-            strk_fee_token_address,
+            chain_info,
             contract_class_mapping_dumper,
         }
     }
 
     pub fn new_for_testing(block_number: BlockNumber) -> Self {
-        RpcStateReader::new(&get_rpc_state_reader_config(), ChainId::Mainnet, block_number, false)
+        RpcStateReader::new(
+            &get_rpc_state_reader_config(),
+            get_chain_info(&ChainId::Mainnet, None),
+            block_number,
+            false,
+        )
     }
 
     // Note: This function is blocking though it is sending a request to the rpc server and waiting
@@ -308,7 +307,7 @@ impl RpcStateReader {
     pub fn get_block_context(&self) -> ReexecutionResult<BlockContext> {
         Ok(BlockContext::new(
             self.get_block_info()?,
-            get_chain_info(&self.chain_id, self.strk_fee_token_address),
+            self.chain_info.clone(),
             self.get_versioned_constants()?,
             BouncerConfig::max(),
         ))
@@ -540,21 +539,21 @@ impl ConsecutiveRpcStateReaders {
     pub fn new(
         last_constructed_block_number: BlockNumber,
         config: Option<RpcStateReaderConfig>,
-        chain_id: ChainId,
+        chain_info: ChainInfo,
         dump_mode: bool,
         contract_class_manager: ContractClassManager,
     ) -> Self {
         let config = config.unwrap_or(get_rpc_state_reader_config());
         Self {
             last_block_state_reader: RpcStateReader::new(
                 &config,
-                chain_id.clone(),
+                chain_info.clone(),
                 last_constructed_block_number,
                 dump_mode,
             ),
             next_block_state_reader: RpcStateReader::new(
                 &config,
-                chain_id,
+                chain_info,
                 last_constructed_block_number.next().expect("Overflow in block number"),
                 dump_mode,
             ),
@@ -603,8 +602,8 @@ impl ConsecutiveRpcStateReaders {
         self,
         tx: Transaction,
     ) -> ReexecutionResult<(TransactionExecutionOutput, StateMaps, BlockContext)> {
-        let chain_id = self.next_block_state_reader.chain_id.clone();
-        let transaction_hash = tx.calculate_transaction_hash(&chain_id)?;
+        let chain_id = &self.next_block_state_reader.chain_info.chain_id;
+        let transaction_hash = tx.calculate_transaction_hash(chain_id)?;
 
         let blockifier_txs = self
             .next_block_state_reader
@@ -640,11 +639,11 @@ impl ConsecutiveRpcStateReaders {
                 let json_content = read_to_string(&tx_path).unwrap_or_else(|_| {
                     panic!("Failed to read transaction JSON file: {}.", tx_path)
                 });
-                let chain_id = self.next_block_state_reader.chain_id.clone();
+                let chain_id = &self.next_block_state_reader.chain_info.chain_id;
 
                 let json_value: Value = serde_json::from_str(&json_content)?;
                 let transaction = deserialize_transaction_json_to_starknet_api_tx(json_value)?;
-                let transaction_hash = transaction.calculate_transaction_hash(&chain_id)?;
+                let transaction_hash = transaction.calculate_transaction_hash(chain_id)?;
 
                 (transaction, transaction_hash)
             }
@@ -659,7 +658,7 @@ impl ConsecutiveRpcStateReaders {
 
     /// Writes the reexecution data required to reexecute a block to a JSON file.
     pub fn write_block_reexecution_data_to_file(self, full_file_path: &str) {
-        let chain_id = self.next_block_state_reader.chain_id.clone();
+        let chain_id = self.next_block_state_reader.chain_info.chain_id.clone();
         let block_number = self.next_block_state_reader.get_block_info().unwrap().block_number;
 
         let serializable_data_next_block = self.get_serializable_data_next_block().unwrap();

crates/blockifier_reexecution/src/state_reader/rpc_state_reader.rs

@@ -16,14 +16,14 @@ use crate::state_reader::rpc_objects::{
     RpcSuccessResponse,
 };
 use crate::state_reader::rpc_state_reader::{RetryConfig, RpcStateReader};
+use crate::utils::get_chain_info;
 
 fn rpc_state_reader_from_latest(config: &RpcStateReaderConfig) -> RpcStateReader {
     RpcStateReader {
         config: config.clone(),
         block_id: BlockId::Latest,
         retry_config: RetryConfig::default(),
-        chain_id: ChainId::Mainnet,
-        strk_fee_token_address: None,
+        chain_info: get_chain_info(&ChainId::Mainnet, None),
         contract_class_mapping_dumper: Arc::new(Mutex::new(None)),
     }
 }