diff --git a/Cargo.lock b/Cargo.lock index 0a636099..466b0bd2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -668,9 +668,9 @@ dependencies = [ [[package]] name = "canhttp" -version = "0.1.1" +version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7ad55f2e8629527f1cd12f9261215a6734363f5859009489ef8647e2124269d9" +checksum = "5b9cf652e717488c8a3ded473a04e577f9446f1d80adc81fbbce0642de786939" dependencies = [ "assert_matches", "ciborium", @@ -678,6 +678,7 @@ dependencies = [ "futures-util", "http 1.3.1", "ic-cdk", + "ic-error-types", "num-traits", "pin-project", "serde", @@ -2108,6 +2109,17 @@ dependencies = [ "zeroize", ] +[[package]] +name = "ic-error-types" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbeeb3d91aa179d6496d7293becdacedfc413c825cac79fd54ea1906f003ee55" +dependencies = [ + "serde", + "strum 0.26.3", + "strum_macros 0.26.4", +] + [[package]] name = "ic-http-types" version = "0.1.0" @@ -4365,6 +4377,7 @@ dependencies = [ "hex", "http 1.3.1", "ic-cdk", + "ic-error-types", "ic-http-types", "ic-metrics-encoder", "ic-stable-structures", @@ -4495,6 +4508,7 @@ dependencies = [ "solana-transaction-status-client-types", "strum 0.27.1", "tokio", + "url", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 4e32f9e3..e0bcbf9a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,7 +25,7 @@ bincode = "1.3.3" bs58 = "0.5.0" candid = "0.10.13" candid_parser = "0.1.4" -canhttp = { version = "0.1.1" } +canhttp = { version = "0.2.0" } canlog = { version = "0.1.0", features = ["derive"] } ciborium = "0.2.2" const_format = "0.2.34" @@ -40,6 +40,7 @@ ic-agent = "0.40.0" ic-canister-log = "0.2.0" ic-cdk = "0.17.2" ic-ed25519 = "0.2.0" +ic-error-types = "0.2" ic-http-types = "0.1.0" ic-management-canister-types = "0.3" ic-metrics-assert = "0.1.0" diff --git a/canister/Cargo.toml b/canister/Cargo.toml index 200e0786..54f16567 100644 --- a/canister/Cargo.toml +++ b/canister/Cargo.toml @@ -30,6 +30,7 @@ derive_more = { workspace = true } hex = { workspace = true } http = { workspace = true } ic-cdk = { workspace = true } +ic-error-types = { workspace = true } ic-http-types = { workspace = true } ic-metrics-encoder = { workspace = true } ic-stable-structures = { workspace = true } diff --git a/canister/src/http/errors.rs b/canister/src/http/errors.rs index d71ff9d4..19d6e63d 100644 --- a/canister/src/http/errors.rs +++ b/canister/src/http/errors.rs @@ -10,6 +10,7 @@ use canhttp::{ CyclesAccountingError, HttpsOutcallError, IcError, }; use derive_more::From; +use ic_error_types::RejectCode; use sol_rpc_types::{HttpOutcallError, ProviderError, RpcError}; use thiserror::Error; @@ -52,6 +53,15 @@ impl From for RpcError { fn from(error: HttpClientError) -> Self { match error { HttpClientError::IcError(IcError { code, message }) => { + use ic_cdk::api::call::RejectionCode as IcCdkRejectionCode; + let code = match code { + RejectCode::SysFatal => IcCdkRejectionCode::SysFatal, + RejectCode::SysTransient => IcCdkRejectionCode::SysTransient, + RejectCode::DestinationInvalid => IcCdkRejectionCode::DestinationInvalid, + RejectCode::CanisterReject => IcCdkRejectionCode::CanisterReject, + RejectCode::CanisterError => IcCdkRejectionCode::CanisterError, + RejectCode::SysUnknown => IcCdkRejectionCode::Unknown, + }; RpcError::HttpOutcallError(HttpOutcallError::IcError { code, message }) } HttpClientError::NotHandledError(e) => RpcError::ValidationError(e), diff --git a/canister/src/http/mod.rs b/canister/src/http/mod.rs index 7f98b9c7..cc96986e 100644 --- a/canister/src/http/mod.rs +++ b/canister/src/http/mod.rs @@ -100,7 +100,7 @@ where .on_error( |req_data: MetricData, error: &HttpClientError| match error { HttpClientError::IcError(IcError { code, message: _ }) => { - observe_response(MetricRpcCallResponse::IcError(*code), &req_data); + observe_response(MetricRpcCallResponse::IcError(code.to_string()), &req_data); } HttpClientError::UnsuccessfulHttpResponse( FilterNonSuccessfulHttpResponseError::UnsuccessfulResponse(response), diff --git a/canister/src/main.rs b/canister/src/main.rs index da8c198b..fbfc7adc 100644 --- a/canister/src/main.rs +++ b/canister/src/main.rs @@ -1,4 +1,5 @@ use candid::candid_method; +use canhttp::multi::Timestamp; use canlog::{log, Log, Sort}; use ic_cdk::{api::is_controller, query, update}; use ic_http_types::{HttpRequest, HttpResponse, HttpResponseBuilder}; @@ -86,7 +87,8 @@ async fn get_account_info( config: Option, params: GetAccountInfoParams, ) -> MultiRpcResult> { - let request = MultiRpcRequest::get_account_info(source, config.unwrap_or_default(), params); + let request = + MultiRpcRequest::get_account_info(source, config.unwrap_or_default(), params, now()); send_multi(request).await.into() } @@ -100,7 +102,7 @@ async fn get_account_info_cycles_cost( if read_state(State::is_demo_mode_active) { return Ok(0); } - MultiRpcRequest::get_account_info(source, config.unwrap_or_default(), params)? + MultiRpcRequest::get_account_info(source, config.unwrap_or_default(), params, now())? .cycles_cost() .await } @@ -112,7 +114,7 @@ async fn get_balance( config: Option, params: GetBalanceParams, ) -> MultiRpcResult { - let request = MultiRpcRequest::get_balance(source, config.unwrap_or_default(), params); + let request = MultiRpcRequest::get_balance(source, config.unwrap_or_default(), params, now()); send_multi(request).await } @@ -126,7 +128,7 @@ async fn get_balance_cycles_cost( if read_state(State::is_demo_mode_active) { return Ok(0); } - MultiRpcRequest::get_balance(source, config.unwrap_or_default(), params)? + MultiRpcRequest::get_balance(source, config.unwrap_or_default(), params, now())? .cycles_cost() .await } @@ -138,7 +140,7 @@ async fn get_block( config: Option, params: GetBlockParams, ) -> MultiRpcResult> { - let request = MultiRpcRequest::get_block(source, config.unwrap_or_default(), params); + let request = MultiRpcRequest::get_block(source, config.unwrap_or_default(), params, now()); send_multi(request).await.into() } @@ -152,7 +154,7 @@ async fn get_block_cycles_cost( if read_state(State::is_demo_mode_active) { return Ok(0); } - MultiRpcRequest::get_block(source, config.unwrap_or_default(), params)? + MultiRpcRequest::get_block(source, config.unwrap_or_default(), params, now())? .cycles_cost() .await } @@ -168,6 +170,7 @@ async fn get_recent_prioritization_fees( source, config.unwrap_or_default(), params.unwrap_or_default(), + now(), ); send_multi(request).await } @@ -186,6 +189,7 @@ async fn get_recent_prioritization_fees_cycles_cost( source, config.unwrap_or_default(), params.unwrap_or_default(), + now(), )? .cycles_cost() .await @@ -198,8 +202,12 @@ async fn get_signatures_for_address( config: Option, params: GetSignaturesForAddressParams, ) -> MultiRpcResult> { - let request = - MultiRpcRequest::get_signatures_for_address(source, config.unwrap_or_default(), params); + let request = MultiRpcRequest::get_signatures_for_address( + source, + config.unwrap_or_default(), + params, + now(), + ); send_multi(request).await } @@ -213,7 +221,7 @@ async fn get_signatures_for_address_cycles_cost( if read_state(State::is_demo_mode_active) { return Ok(0); } - MultiRpcRequest::get_signatures_for_address(source, config.unwrap_or_default(), params)? + MultiRpcRequest::get_signatures_for_address(source, config.unwrap_or_default(), params, now())? .cycles_cost() .await } @@ -226,7 +234,7 @@ async fn get_signature_statuses( params: GetSignatureStatusesParams, ) -> MultiRpcResult>> { let request = - MultiRpcRequest::get_signature_statuses(source, config.unwrap_or_default(), params); + MultiRpcRequest::get_signature_statuses(source, config.unwrap_or_default(), params, now()); send_multi(request).await.into() } @@ -240,7 +248,7 @@ async fn get_signature_statuses_cycles_cost( if read_state(State::is_demo_mode_active) { return Ok(0); } - MultiRpcRequest::get_signature_statuses(source, config.unwrap_or_default(), params)? + MultiRpcRequest::get_signature_statuses(source, config.unwrap_or_default(), params, now())? .cycles_cost() .await } @@ -256,6 +264,7 @@ async fn get_slot( source, config.unwrap_or_default(), params.unwrap_or_default(), + now(), ); send_multi(request).await } @@ -274,6 +283,7 @@ async fn get_slot_cycles_cost( source, config.unwrap_or_default(), params.unwrap_or_default(), + now(), )? .cycles_cost() .await @@ -286,8 +296,12 @@ async fn get_token_account_balance( config: Option, params: GetTokenAccountBalanceParams, ) -> MultiRpcResult { - let request = - MultiRpcRequest::get_token_account_balance(source, config.unwrap_or_default(), params); + let request = MultiRpcRequest::get_token_account_balance( + source, + config.unwrap_or_default(), + params, + now(), + ); send_multi(request).await.into() } @@ -301,7 +315,7 @@ async fn get_token_account_balance_cycles_cost( if read_state(State::is_demo_mode_active) { return Ok(0); } - MultiRpcRequest::get_token_account_balance(source, config.unwrap_or_default(), params)? + MultiRpcRequest::get_token_account_balance(source, config.unwrap_or_default(), params, now())? .cycles_cost() .await } @@ -313,7 +327,8 @@ async fn get_transaction( config: Option, params: GetTransactionParams, ) -> MultiRpcResult> { - let request = MultiRpcRequest::get_transaction(source, config.unwrap_or_default(), params); + let request = + MultiRpcRequest::get_transaction(source, config.unwrap_or_default(), params, now()); send_multi(request).await.into() } @@ -327,7 +342,7 @@ async fn get_transaction_cycles_cost( if read_state(State::is_demo_mode_active) { return Ok(0); } - MultiRpcRequest::get_transaction(source, config.unwrap_or_default(), params)? + MultiRpcRequest::get_transaction(source, config.unwrap_or_default(), params, now())? .cycles_cost() .await } @@ -339,7 +354,8 @@ async fn send_transaction( config: Option, params: SendTransactionParams, ) -> MultiRpcResult { - let request = MultiRpcRequest::send_transaction(source, config.unwrap_or_default(), params); + let request = + MultiRpcRequest::send_transaction(source, config.unwrap_or_default(), params, now()); send_multi(request).await } @@ -353,7 +369,7 @@ async fn send_transaction_cycles_cost( if read_state(State::is_demo_mode_active) { return Ok(0); } - MultiRpcRequest::send_transaction(source, config.unwrap_or_default(), params)? + MultiRpcRequest::send_transaction(source, config.unwrap_or_default(), params, now())? .cycles_cost() .await } @@ -366,7 +382,7 @@ async fn json_request( json_rpc_payload: String, ) -> MultiRpcResult { let request = - MultiRpcRequest::json_request(source, config.unwrap_or_default(), json_rpc_payload); + MultiRpcRequest::json_request(source, config.unwrap_or_default(), json_rpc_payload, now()); send_multi(request).await.map(|value| value.to_string()) } @@ -380,7 +396,7 @@ async fn json_request_cycles_cost( if read_state(State::is_demo_mode_active) { return Ok(0); } - MultiRpcRequest::json_request(source, config.unwrap_or_default(), json_rpc_payload)? + MultiRpcRequest::json_request(source, config.unwrap_or_default(), json_rpc_payload, now())? .cycles_cost() .await } @@ -483,6 +499,10 @@ fn post_upgrade(args: Option) { lifecycle::post_upgrade(args); } +fn now() -> Timestamp { + Timestamp::from_nanos_since_unix_epoch(ic_cdk::api::time()) +} + fn main() {} #[test] diff --git a/canister/src/memory/mod.rs b/canister/src/memory/mod.rs index 28fb85d3..ea7d44de 100644 --- a/canister/src/memory/mod.rs +++ b/canister/src/memory/mod.rs @@ -1,12 +1,14 @@ #[cfg(test)] mod tests; +use crate::providers::SupportedRpcProviderUsage; use crate::{ metrics::Metrics, types::{ApiKey, OverrideProvider}, }; use candid::{Deserialize, Principal}; use canhttp::http::json::{ConstantSizeId, Id}; +use canhttp::multi::Timestamp; use canlog::LogFilter; use ic_stable_structures::{ memory_manager::{MemoryId, MemoryManager, VirtualMemory}, @@ -25,6 +27,7 @@ thread_local! { // Unstable static data: these are reset when the canister is upgraded. pub static UNSTABLE_METRICS: RefCell = RefCell::new(Metrics::default()); static UNSTABLE_HTTP_REQUEST_COUNTER: RefCell = const {RefCell::new(ConstantSizeId::ZERO)}; + static UNSTABLE_RPC_SERVICE_OK_RESULTS_TIMESTAMPS: RefCell = RefCell::new(SupportedRpcProviderUsage::default()); // Stable static data: these are preserved when the canister is upgraded. static MEMORY_MANAGER: RefCell> = @@ -226,3 +229,16 @@ pub fn next_request_id() -> Id { Id::from(current_request_id) }) } + +pub fn record_ok_result(provider: SupportedRpcProviderId, now: Timestamp) { + UNSTABLE_RPC_SERVICE_OK_RESULTS_TIMESTAMPS + .with_borrow_mut(|access| access.record_evict(provider, now)); +} + +pub fn rank_providers( + providers: &[SupportedRpcProviderId], + now: Timestamp, +) -> Vec { + UNSTABLE_RPC_SERVICE_OK_RESULTS_TIMESTAMPS + .with_borrow_mut(|access| access.rank_ascending_evict(providers, now)) +} diff --git a/canister/src/metrics/mod.rs b/canister/src/metrics/mod.rs index ca4a885c..cf60fe82 100644 --- a/canister/src/metrics/mod.rs +++ b/canister/src/metrics/mod.rs @@ -1,5 +1,4 @@ use derive_more::From; -use ic_cdk::api::call::RejectionCode; use std::collections::BTreeMap; use std::time::Duration; @@ -195,21 +194,6 @@ impl MetricLabels for MetricHttpStatusCode { } } -impl MetricLabels for RejectionCode { - fn metric_labels(&self) -> Vec<(&str, &str)> { - let code = match self { - RejectionCode::NoError => "NO_ERROR", - RejectionCode::SysFatal => "SYS_FATAL", - RejectionCode::SysTransient => "SYS_TRANSIENT", - RejectionCode::DestinationInvalid => "DESTINATION_INVALID", - RejectionCode::CanisterReject => "CANISTER_REJECT", - RejectionCode::CanisterError => "CANISTER_ERROR", - RejectionCode::Unknown => "UNKNOWN", - }; - vec![("code", code)] - } -} - #[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, From)] pub struct MetricRpcErrorCode(pub String); @@ -228,7 +212,7 @@ impl MetricLabels for MetricRpcErrorCode { #[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)] pub enum MetricRpcCallResponse { Success, - IcError(RejectionCode), + IcError(String), HttpError(MetricHttpStatusCode), JsonRpcError, } @@ -239,7 +223,7 @@ impl MetricLabels for MetricRpcCallResponse { MetricRpcCallResponse::Success => vec![], MetricRpcCallResponse::IcError(rejection_code) => [("error", "ic")] .into_iter() - .chain(rejection_code.metric_labels()) + .chain(vec![("code", rejection_code.as_str())]) .collect(), MetricRpcCallResponse::HttpError(status) => [("error", "http")] .into_iter() diff --git a/canister/src/providers/mod.rs b/canister/src/providers/mod.rs index d8ec0d82..4ba9b572 100644 --- a/canister/src/providers/mod.rs +++ b/canister/src/providers/mod.rs @@ -1,7 +1,9 @@ #[cfg(test)] mod tests; +use crate::memory::rank_providers; use crate::{constants::API_KEY_REPLACE_STRING, memory::read_state, types::OverrideProvider}; +use canhttp::multi::{TimedSizedMap, TimedSizedVec, Timestamp}; use ic_cdk::api::management_canister::http_request::HttpHeader; use maplit::btreemap; use sol_rpc_types::{ @@ -9,6 +11,8 @@ use sol_rpc_types::{ RpcSource, RpcSources, SolanaCluster, SupportedRpcProvider, SupportedRpcProviderId, }; use std::collections::{BTreeMap, BTreeSet}; +use std::num::NonZeroUsize; +use std::time::Duration; thread_local! { pub static PROVIDERS: BTreeMap = btreemap! { @@ -144,7 +148,11 @@ impl Providers { const DEFAULT_NUM_PROVIDERS_FOR_EQUALITY: usize = 3; - pub fn new(source: RpcSources, strategy: ConsensusStrategy) -> Result { + pub fn new( + source: RpcSources, + strategy: ConsensusStrategy, + now: Timestamp, + ) -> Result { fn supported_providers( cluster: &SolanaCluster, ) -> Result<&[SupportedRpcProviderId], ProviderError> { @@ -157,8 +165,8 @@ impl Providers { } } - fn supported_rpc_source(supported_provider: &SupportedRpcProviderId) -> RpcSource { - RpcSource::Supported(*supported_provider) + fn supported_rpc_source(supported_provider: SupportedRpcProviderId) -> RpcSource { + RpcSource::Supported(supported_provider) } let providers: BTreeSet<_> = match strategy { @@ -170,8 +178,8 @@ impl Providers { supported_providers.len() >= Self::DEFAULT_NUM_PROVIDERS_FOR_EQUALITY, "BUG: need at least 3 providers, but got {supported_providers:?}" ); - Ok(supported_providers - .iter() + Ok(rank_providers(supported_providers, now) + .into_iter() .take(Self::DEFAULT_NUM_PROVIDERS_FOR_EQUALITY) .map(supported_rpc_source) .collect()) @@ -227,8 +235,8 @@ impl Providers { total, all_providers_len ))); } - let providers: BTreeSet<_> = supported_providers - .iter() + let providers: BTreeSet<_> = rank_providers(supported_providers, now) + .into_iter() .take(total as usize) .map(supported_rpc_source) .collect(); @@ -306,3 +314,42 @@ pub fn request_builder( } Ok(request_builder) } + +/// Record when a supported RPC service was used. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SupportedRpcProviderUsage(TimedSizedMap); + +impl Default for SupportedRpcProviderUsage { + fn default() -> Self { + Self::new() + } +} + +impl SupportedRpcProviderUsage { + pub fn new() -> SupportedRpcProviderUsage { + Self(TimedSizedMap::new( + Duration::from_secs(20 * 60), + NonZeroUsize::new(500).unwrap(), + )) + } + + pub fn record_evict(&mut self, service: SupportedRpcProviderId, now: Timestamp) { + self.0.insert_evict(now, service, ()); + } + + pub fn rank_ascending_evict( + &mut self, + providers: &[SupportedRpcProviderId], + now: Timestamp, + ) -> Vec { + fn ascending_num_elements(values: Option<&TimedSizedVec>) -> impl Ord { + std::cmp::Reverse(values.map(|v| v.len()).unwrap_or_default()) + } + + self.0.evict_expired(providers, now); + self.0 + .sort_keys_by(providers, ascending_num_elements) + .copied() + .collect() + } +} diff --git a/canister/src/providers/tests.rs b/canister/src/providers/tests.rs index a3e8f7c3..f5ed50d4 100644 --- a/canister/src/providers/tests.rs +++ b/canister/src/providers/tests.rs @@ -73,6 +73,7 @@ fn should_partition_providers_between_solana_cluster() { mod providers_new { use crate::providers::Providers; use assert_matches::assert_matches; + use canhttp::multi::Timestamp; use maplit::btreeset; use sol_rpc_types::{ ConsensusStrategy, ProviderError, RpcSource, RpcSources, SolanaCluster, @@ -82,7 +83,11 @@ mod providers_new { #[test] fn should_fail_when_providers_explicitly_set_to_empty() { assert_matches!( - Providers::new(RpcSources::Custom(vec![]), ConsensusStrategy::default()), + Providers::new( + RpcSources::Custom(vec![]), + ConsensusStrategy::default(), + Timestamp::UNIX_EPOCH + ), Err(ProviderError::InvalidRpcConfig(_)) ); } @@ -90,8 +95,12 @@ mod providers_new { #[test] fn should_use_default_providers() { for cluster in [SolanaCluster::Mainnet, SolanaCluster::Devnet] { - let providers = - Providers::new(RpcSources::Default(cluster), ConsensusStrategy::default()).unwrap(); + let providers = Providers::new( + RpcSources::Default(cluster), + ConsensusStrategy::default(), + Timestamp::UNIX_EPOCH, + ) + .unwrap(); assert!(!providers.sources.is_empty()); } } @@ -107,6 +116,7 @@ mod providers_new { RpcSource::Supported(provider2), ]), ConsensusStrategy::default(), + Timestamp::UNIX_EPOCH, ) .unwrap(); @@ -119,3 +129,82 @@ mod providers_new { ); } } + +mod supported_rpc_provider_usage { + use crate::providers::{Providers, SupportedRpcProviderUsage}; + use canhttp::multi::Timestamp; + use sol_rpc_types::{SolanaCluster, SupportedRpcProviderId}; + use std::time::Duration; + + const MINUTE: Duration = Duration::from_secs(60); + + #[test] + fn should_have_default_ordering_when_no_data() { + let mut usage = SupportedRpcProviderUsage::default(); + + for (_cluster, providers) in all_supported_providers() { + let ordered = usage.rank_ascending_evict(providers, Timestamp::UNIX_EPOCH); + assert_eq!(ordered, providers); + } + } + + #[test] + fn should_have_default_ordering_when_data_expired() { + let mut usage = SupportedRpcProviderUsage::default(); + let now = Timestamp::UNIX_EPOCH; + for (_cluster, supported_providers) in all_supported_providers() { + let last_provider = *supported_providers.last().unwrap(); + usage.record_evict(last_provider, now); + } + + let expired = Timestamp::from_unix_epoch(21 * MINUTE); + for (_cluster, supported_providers) in all_supported_providers() { + let ordered = usage.rank_ascending_evict(supported_providers, expired); + assert_eq!(ordered, supported_providers); + } + } + + #[test] + fn should_rank_based_on_non_expired_data() { + let mut usage = SupportedRpcProviderUsage::default(); + for (_cluster, supported_providers) in all_supported_providers() { + assert!(supported_providers.len() >= 2); + + // 3 entries, 2 expire after > 20 minutes + usage.record_evict(supported_providers[0], Timestamp::UNIX_EPOCH); + usage.record_evict(supported_providers[0], Timestamp::UNIX_EPOCH); + usage.record_evict(supported_providers[0], Timestamp::from_unix_epoch(MINUTE)); + + // 3 entries, 1 expire after > 20 minutes + usage.record_evict(supported_providers[1], Timestamp::UNIX_EPOCH); + usage.record_evict(supported_providers[1], Timestamp::from_unix_epoch(MINUTE)); + usage.record_evict(supported_providers[1], Timestamp::from_unix_epoch(MINUTE)); + } + + for (_cluster, supported_providers) in all_supported_providers() { + let non_expired = Timestamp::from_unix_epoch(20 * MINUTE); + let usage_before = usage.clone(); + let ordered = usage.rank_ascending_evict(supported_providers, non_expired); + assert_eq!(ordered, supported_providers); + assert_eq!(usage, usage_before); + + let expired = Timestamp::from_unix_epoch(21 * MINUTE); + let usage_before = usage.clone(); + let ordered = usage.rank_ascending_evict(supported_providers, expired); + let expected_order = { + let mut expected = vec![supported_providers[1], supported_providers[0]]; + expected.extend(&supported_providers[2..]); + expected + }; + assert_eq!(ordered, expected_order); + assert_ne!(usage, usage_before); + } + } + + fn all_supported_providers() -> [(SolanaCluster, &'static [SupportedRpcProviderId]); 2] { + [ + (SolanaCluster::Mainnet, Providers::MAINNET_PROVIDERS), + (SolanaCluster::Devnet, Providers::DEVNET_PROVIDERS), + ] + } +} diff --git a/canister/src/rpc_client/mod.rs b/canister/src/rpc_client/mod.rs index d3d38135..a7e724a0 100644 --- a/canister/src/rpc_client/mod.rs +++ b/canister/src/rpc_client/mod.rs @@ -4,6 +4,7 @@ mod sol_rpc; #[cfg(test)] mod tests; +use crate::memory::record_ok_result; use crate::{ http::{ errors::HttpClientError, http_client, service_request_builder, ChargingPolicyWithCollateral, @@ -13,6 +14,7 @@ use crate::{ providers::{request_builder, resolve_rpc_provider, Providers}, rpc_client::sol_rpc::ResponseTransform, }; +use canhttp::multi::Timestamp; use canhttp::{ http::json::JsonRpcRequest, multi::{MultiResults, Reduce, ReduceWithEquality, ReduceWithThreshold}, @@ -95,9 +97,10 @@ impl GetAccountInfoRequest { rpc_sources: RpcSources, config: RpcConfig, params: Params, + now: Timestamp, ) -> Result { let consensus_strategy = config.response_consensus.unwrap_or_default(); - let providers = Providers::new(rpc_sources, consensus_strategy.clone())?; + let providers = Providers::new(rpc_sources, consensus_strategy.clone(), now)?; let max_response_bytes = config .response_size_estimate .unwrap_or(512 + HEADER_SIZE_LIMIT); @@ -119,9 +122,10 @@ impl GetBalanceRequest { rpc_sources: RpcSources, config: RpcConfig, params: Params, + now: Timestamp, ) -> Result { let consensus_strategy = config.response_consensus.unwrap_or_default(); - let providers = Providers::new(rpc_sources, consensus_strategy.clone())?; + let providers = Providers::new(rpc_sources, consensus_strategy.clone(), now)?; let max_response_bytes = config .response_size_estimate .unwrap_or(256 + HEADER_SIZE_LIMIT); @@ -146,10 +150,11 @@ impl GetBlockRequest { rpc_sources: RpcSources, config: RpcConfig, params: Params, + now: Timestamp, ) -> Result { let params = params.into(); let consensus_strategy = config.response_consensus.unwrap_or_default(); - let providers = Providers::new(rpc_sources, consensus_strategy.clone())?; + let providers = Providers::new(rpc_sources, consensus_strategy.clone(), now)?; let max_response_bytes = Self::response_size_estimate(¶ms); Ok(MultiRpcRequest::new( @@ -190,10 +195,11 @@ impl GetSignaturesForAddressRequest { rpc_sources: RpcSources, config: RpcConfig, params: Params, + now: Timestamp, ) -> Result { let params = params.into(); let consensus_strategy = config.response_consensus.unwrap_or_default(); - let providers = Providers::new(rpc_sources, consensus_strategy.clone())?; + let providers = Providers::new(rpc_sources, consensus_strategy.clone(), now)?; let max_response_bytes = config .response_size_estimate .unwrap_or((params.get_limit() as u64 * 256) + HEADER_SIZE_LIMIT); @@ -218,10 +224,11 @@ impl GetSignatureStatusesRequest { rpc_sources: RpcSources, config: RpcConfig, params: Params, + now: Timestamp, ) -> Result { let params = params.into(); let consensus_strategy = config.response_consensus.unwrap_or_default(); - let providers = Providers::new(rpc_sources, consensus_strategy.clone())?; + let providers = Providers::new(rpc_sources, consensus_strategy.clone(), now)?; let max_response_bytes = config .response_size_estimate .unwrap_or(128 + (params.num_signatures() as u64 * 256) + HEADER_SIZE_LIMIT); @@ -243,9 +250,10 @@ impl GetSlotRequest { rpc_sources: RpcSources, config: GetSlotRpcConfig, params: Params, + now: Timestamp, ) -> Result { let consensus_strategy = config.response_consensus.unwrap_or_default(); - let providers = Providers::new(rpc_sources, consensus_strategy.clone())?; + let providers = Providers::new(rpc_sources, consensus_strategy.clone(), now)?; let max_response_bytes = config .response_size_estimate .unwrap_or(64 + HEADER_SIZE_LIMIT); @@ -269,10 +277,11 @@ impl GetRecentPrioritizationFeesRequest { rpc_sources: RpcSources, config: GetRecentPrioritizationFeesRpcConfig, params: Params, + now: Timestamp, ) -> Result { let max_length = config.max_length(); let consensus_strategy = config.response_consensus.unwrap_or_default(); - let providers = Providers::new(rpc_sources, consensus_strategy.clone())?; + let providers = Providers::new(rpc_sources, consensus_strategy.clone(), now)?; let max_response_bytes = config .response_size_estimate .unwrap_or(8 * 1024 + HEADER_SIZE_LIMIT); @@ -300,9 +309,10 @@ impl GetTokenAccountBalanceRequest { rpc_sources: RpcSources, config: RpcConfig, params: Params, + now: Timestamp, ) -> Result { let consensus_strategy = config.response_consensus.unwrap_or_default(); - let providers = Providers::new(rpc_sources, consensus_strategy.clone())?; + let providers = Providers::new(rpc_sources, consensus_strategy.clone(), now)?; let max_response_bytes = config .response_size_estimate .unwrap_or(256 + HEADER_SIZE_LIMIT); @@ -327,9 +337,10 @@ impl GetTransactionRequest { rpc_sources: RpcSources, config: RpcConfig, params: Params, + now: Timestamp, ) -> Result { let consensus_strategy = config.response_consensus.unwrap_or_default(); - let providers = Providers::new(rpc_sources, consensus_strategy.clone())?; + let providers = Providers::new(rpc_sources, consensus_strategy.clone(), now)?; let max_response_bytes = config .response_size_estimate .unwrap_or(8 * 1024 + HEADER_SIZE_LIMIT); @@ -351,9 +362,10 @@ impl SendTransactionRequest { rpc_sources: RpcSources, config: RpcConfig, params: Params, + now: Timestamp, ) -> Result { let consensus_strategy = config.response_consensus.unwrap_or_default(); - let providers = Providers::new(rpc_sources, consensus_strategy.clone())?; + let providers = Providers::new(rpc_sources, consensus_strategy.clone(), now)?; let max_response_bytes = config .response_size_estimate .unwrap_or(128 + HEADER_SIZE_LIMIT); @@ -375,6 +387,7 @@ impl JsonRequest { rpc_sources: RpcSources, config: RpcConfig, json_rpc_payload: String, + now: Timestamp, ) -> RpcResult { let request: JsonRpcRequest = match serde_json::from_str(&json_rpc_payload) { @@ -386,7 +399,7 @@ impl JsonRequest { } }; let consensus_strategy = config.response_consensus.unwrap_or_default(); - let providers = Providers::new(rpc_sources, consensus_strategy.clone())?; + let providers = Providers::new(rpc_sources, consensus_strategy.clone(), now)?; let max_response_bytes = config .response_size_estimate .unwrap_or(1024 + HEADER_SIZE_LIMIT); @@ -435,6 +448,12 @@ impl MultiRpcRequest { let (requests, errors) = requests.into_inner(); let (_client, mut results) = canhttp::multi::parallel_call(client, requests).await; results.add_errors(errors); + let now = Timestamp::from_nanos_since_unix_epoch(ic_cdk::api::time()); + results + .ok_results() + .keys() + .filter_map(RpcSource::rpc_provider_id) + .for_each(|provider_id| record_ok_result(provider_id, now)); assert_eq!( results.len(), num_providers, diff --git a/canister/src/rpc_client/tests.rs b/canister/src/rpc_client/tests.rs index bddd8904..ef9885b8 100644 --- a/canister/src/rpc_client/tests.rs +++ b/canister/src/rpc_client/tests.rs @@ -24,6 +24,7 @@ const ANOTHER_SIGNATURE: &str = mod request_serialization_tests { use super::*; + use canhttp::multi::Timestamp; #[test] fn should_serialize_get_account_info_request() { @@ -32,6 +33,7 @@ mod request_serialization_tests { RpcSources::Default(SolanaCluster::Mainnet), RpcConfig::default(), GetAccountInfoParams::from(solana_pubkey::Pubkey::default()), + Timestamp::default(), ) .unwrap(), json!(["11111111111111111111111111111111", null]), @@ -50,6 +52,7 @@ mod request_serialization_tests { }), min_context_slot: Some(456), }, + Timestamp::default(), ) .unwrap(), json!([ @@ -70,6 +73,7 @@ mod request_serialization_tests { RpcSources::Default(SolanaCluster::Mainnet), GetSlotRpcConfig::default(), GetSlotParams::default(), + Timestamp::default(), ) .unwrap(), json!([null]), @@ -82,6 +86,7 @@ mod request_serialization_tests { commitment: Some(CommitmentLevel::Finalized), min_context_slot: Some(123), }, + Timestamp::default(), ) .unwrap(), json!([ @@ -107,6 +112,7 @@ mod request_serialization_tests { before: None, until: None, }, + Timestamp::default(), ) .unwrap(), json!(["11111111111111111111111111111111", null]), @@ -123,6 +129,7 @@ mod request_serialization_tests { before: Some(Signature::from_str(SOME_SIGNATURE).unwrap()), until: Some(Signature::from_str(ANOTHER_SIGNATURE).unwrap()), }, + Timestamp::default(), ) .unwrap(), json!([ @@ -148,6 +155,7 @@ mod request_serialization_tests { signatures: VecWithMaxLen::new(), search_transaction_history: None, }, + Timestamp::default(), ) .unwrap(), json!([[], null]), @@ -165,6 +173,7 @@ mod request_serialization_tests { .unwrap(), search_transaction_history: Some(true), }, + Timestamp::default(), ) .unwrap(), json!([ @@ -184,6 +193,7 @@ mod request_serialization_tests { RpcSources::Default(SolanaCluster::Mainnet), RpcConfig::default(), GetTransactionParams::from(solana_signature::Signature::default()), + Timestamp::default(), ) .unwrap(), json!([signature, null]), @@ -198,6 +208,7 @@ mod request_serialization_tests { max_supported_transaction_version: Some(2), encoding: Some(GetTransactionEncoding::Base64), }, + Timestamp::default(), ) .unwrap(), json!([ @@ -219,6 +230,7 @@ mod request_serialization_tests { RpcSources::Default(SolanaCluster::Mainnet), RpcConfig::default(), GetBalanceParams::from(pubkey), + Timestamp::default(), ) .unwrap(), json!([pubkey.to_string(), null]), @@ -233,6 +245,7 @@ mod request_serialization_tests { commitment: Some(CommitmentLevel::Confirmed), min_context_slot: Some(42), }, + Timestamp::default(), ) .unwrap(), json!( @@ -255,6 +268,7 @@ mod request_serialization_tests { RpcSources::Default(SolanaCluster::Mainnet), RpcConfig::default(), GetTokenAccountBalanceParams::from(pubkey), + Timestamp::default(), ) .unwrap(), json!([pubkey.to_string(), null]), @@ -268,6 +282,7 @@ mod request_serialization_tests { pubkey: pubkey.into(), commitment: Some(CommitmentLevel::Confirmed), }, + Timestamp::default(), ) .unwrap(), json!([ @@ -284,6 +299,7 @@ mod request_serialization_tests { RpcSources::Default(SolanaCluster::Mainnet), RpcConfig::default(), GetBlockParams::from(123), + Timestamp::default(), ) .unwrap(), json!([123, {"transactionDetails": "none"}]), @@ -299,6 +315,7 @@ mod request_serialization_tests { transaction_details: Some(TransactionDetails::Signatures), rewards: Some(true), }, + Timestamp::default(), ) .unwrap(), json!([ @@ -320,6 +337,7 @@ mod request_serialization_tests { RpcSources::Default(SolanaCluster::Mainnet), GetRecentPrioritizationFeesRpcConfig::default(), GetRecentPrioritizationFeesParams::default(), + Timestamp::default(), ) .unwrap(), json!([[]]), @@ -334,6 +352,7 @@ mod request_serialization_tests { pubkey!("3emsAVdmGKERbHjmGfQ6oZ1e35dkf5iYcS6U4CPKFVaa"), ]) .unwrap(), + Timestamp::default(), ) .unwrap(), json!([[ @@ -354,6 +373,7 @@ mod request_serialization_tests { transaction.to_string(), SendTransactionEncoding::Base64, ), + Timestamp::default(), ) .unwrap(), json!([transaction, { "encoding": "base64" }]), @@ -371,6 +391,7 @@ mod request_serialization_tests { RpcSources::Default(SolanaCluster::Mainnet), RpcConfig::default(), params, + Timestamp::default(), ) .unwrap(), json!([ diff --git a/integration_tests/Cargo.toml b/integration_tests/Cargo.toml index 76cf6669..8e16bbd9 100644 --- a/integration_tests/Cargo.toml +++ b/integration_tests/Cargo.toml @@ -40,6 +40,7 @@ solana-signer = { workspace = true } solana-transaction = { workspace = true } solana-transaction-status-client-types = { workspace = true } tokio = { workspace = true } +url = { workspace = true } [dev-dependencies] assert_matches = { workspace = true } diff --git a/integration_tests/src/mock.rs b/integration_tests/src/mock.rs index e6143f90..b9cda8eb 100644 --- a/integration_tests/src/mock.rs +++ b/integration_tests/src/mock.rs @@ -6,6 +6,8 @@ use pocket_ic::common::rest::{ }; use serde_json::Value; use std::collections::BTreeSet; +use std::str::FromStr; +use url::{Host, Url}; pub struct MockOutcallBody(pub Vec); @@ -43,6 +45,7 @@ impl MockOutcallBuilder { Self(MockOutcall { method: None, url: None, + host: None, request_headers: None, request_body: None, max_response_bytes: None, @@ -58,6 +61,7 @@ impl MockOutcallBuilder { Self(MockOutcall { method: None, url: None, + host: None, request_headers: None, request_body: None, max_response_bytes: None, @@ -78,6 +82,11 @@ impl MockOutcallBuilder { self } + pub fn with_host(mut self, host: &str) -> Self { + self.0.host = Some(Host::parse(host).expect("BUG: invalid host for a URL")); + self + } + pub fn with_request_headers(mut self, headers: Vec<(impl ToString, impl ToString)>) -> Self { self.0.request_headers = Some( headers @@ -120,6 +129,7 @@ impl From for MockOutcall { pub struct MockOutcall { pub method: Option, pub url: Option, + pub host: Option, pub request_headers: Option>, pub request_body: Option>, pub max_response_bytes: Option, @@ -128,8 +138,16 @@ pub struct MockOutcall { impl MockOutcall { pub fn assert_matches(&self, request: &CanisterHttpRequest) { + let req_url = Url::from_str(&request.url).expect("BUG: invalid URL"); if let Some(ref url) = self.url { - assert_eq!(url, &request.url); + let mock_url = Url::from_str(url).unwrap(); + assert_eq!(mock_url, req_url); + } + if let Some(ref host) = self.host { + assert_eq!( + host, + &req_url.host().expect("BUG: missing host in URL").to_owned() + ); } if let Some(ref method) = self.method { assert_eq!(method, &request.http_method); diff --git a/integration_tests/tests/tests.rs b/integration_tests/tests/tests.rs index 43c80ce6..bd0a6ced 100644 --- a/integration_tests/tests/tests.rs +++ b/integration_tests/tests/tests.rs @@ -13,10 +13,10 @@ use sol_rpc_int_tests::{ mock::MockOutcallBuilder, PocketIcRuntime, Setup, SolRpcTestClient, DEFAULT_CALLER_TEST_ID, }; use sol_rpc_types::{ - CommitmentLevel, ConfirmedTransactionStatusWithSignature, GetSignaturesForAddressLimit, - GetSlotParams, InstallArgs, InstructionError, Mode, ProviderError, RpcAccess, RpcAuth, - RpcEndpoint, RpcError, RpcResult, RpcSource, RpcSources, Slot, SolanaCluster, - SupportedRpcProvider, SupportedRpcProviderId, TransactionError, + CommitmentLevel, ConfirmedTransactionStatusWithSignature, ConsensusStrategy, + GetSignaturesForAddressLimit, GetSlotParams, InstallArgs, InstructionError, Mode, + ProviderError, RpcAccess, RpcAuth, RpcEndpoint, RpcError, RpcResult, RpcSource, RpcSources, + Slot, SolanaCluster, SupportedRpcProvider, SupportedRpcProviderId, TransactionError, }; use solana_account_decoder_client_types::{ token::UiTokenAmount, UiAccount, UiAccountData, UiAccountEncoding, @@ -2232,6 +2232,86 @@ async fn should_log_request_and_response() { setup.drop().await; } +#[tokio::test] +async fn should_change_default_providers_when_one_keeps_failing() { + fn request_body(id: u8) -> serde_json::Value { + let id = ConstantSizeId::from(id).to_string(); + json!({ "jsonrpc": "2.0", "id": id, "method": "getSlot", "params": [null] }) + } + + fn response_body(id: u8) -> serde_json::Value { + let id = ConstantSizeId::from(id).to_string(); + json!({ "id": id, "jsonrpc": "2.0", "result": 1200, }) + } + let setup = Setup::new().await.with_mock_api_keys().await; + + let client = setup.client(); + let slot = client + .with_consensus_strategy(ConsensusStrategy::Threshold { + min: 2, + total: Some(3), + }) + .mock_http_sequence(vec![ + MockOutcallBuilder::new(200, response_body(0)) + .with_request_body(request_body(0)) + .with_host("solana-mainnet.g.alchemy.com"), + MockOutcallBuilder::new(500, "error") + .with_request_body(request_body(1)) + .with_host("lb.drpc.org"), + MockOutcallBuilder::new(200, response_body(2)) + .with_request_body(request_body(2)) + .with_host("mainnet.helius-rpc.com"), + ]) + .build() + .get_slot() + .send() + .await + .expect_consistent(); + assert_eq!(slot, Ok(1200)); + + let client = setup.client(); + let slot = client + .with_consensus_strategy(ConsensusStrategy::Equality) + .with_rpc_sources(RpcSources::Custom(vec![RpcSource::Supported( + SupportedRpcProviderId::AnkrMainnet, + )])) + .mock_http_sequence(vec![MockOutcallBuilder::new(200, response_body(3)) + .with_request_body(request_body(3)) + .with_host("rpc.ankr.com")]) + .build() + .get_slot() + .send() + .await + .expect_consistent(); + assert_eq!(slot, Ok(1200)); + + let client = setup.client(); + let slot = client + .with_consensus_strategy(ConsensusStrategy::Threshold { + min: 3, + total: Some(3), + }) + .mock_http_sequence(vec![ + MockOutcallBuilder::new(200, response_body(4)) + .with_request_body(request_body(4)) + .with_host("solana-mainnet.g.alchemy.com"), + MockOutcallBuilder::new(200, response_body(5)) + .with_request_body(request_body(5)) + .with_host("rpc.ankr.com"), + MockOutcallBuilder::new(200, response_body(6)) + .with_request_body(request_body(6)) + .with_host("mainnet.helius-rpc.com"), + ]) + .build() + .get_slot() + .send() + .await + .expect_consistent(); + assert_eq!(slot, Ok(1200)); + + setup.drop().await; +} + fn assert_within(actual: u128, expected: u128, percentage_error: u8) { assert!(percentage_error <= 100); let error_margin = expected.saturating_mul(percentage_error as u128) / 100; diff --git a/libs/types/src/rpc_client/mod.rs b/libs/types/src/rpc_client/mod.rs index e9fb00bc..e693136a 100644 --- a/libs/types/src/rpc_client/mod.rs +++ b/libs/types/src/rpc_client/mod.rs @@ -85,7 +85,8 @@ pub enum HttpOutcallError { /// Response is not a valid JSON-RPC response, /// which means that the response was not successful (status other than 2xx) /// or that the response body could not be deserialized into a JSON-RPC response. - #[error("Invalid HTTP JSON-RPC response: status {status}, body: {body}, parsing error: {parsing_error:?}")] + #[error("Invalid HTTP JSON-RPC response: status {status}, body: {body}, parsing error: {parsing_error:?}" + )] InvalidHttpJsonRpcResponse { /// The HTTP status code returned. status: u16, @@ -373,6 +374,17 @@ pub enum RpcSource { Custom(RpcEndpoint), } +impl RpcSource { + /// Return the supported RPC provider ID if the source corresponds to a supported provider, + /// `None` otherwise. + pub fn rpc_provider_id(&self) -> Option { + match &self { + RpcSource::Supported(id) => Some(*id), + RpcSource::Custom(_) => None, + } + } +} + /// Defines a collection of Solana RPC sources. #[derive( Clone, PartialEq, Eq, Ord, PartialOrd, Hash, Serialize, Deserialize, CandidType, Debug,