diff --git a/Cargo.lock b/Cargo.lock index 8121c66fa594..117d978d20f1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10317,6 +10317,7 @@ dependencies = [ "ic-test-utilities-time", "ic-test-utilities-types", "ic-types", + "ic-utils 0.9.0", "prometheus", "slog", "tokio", @@ -10363,7 +10364,6 @@ dependencies = [ "ic-test-utilities-types", "ic-types", "ic-types-cycles", - "ic-utils 0.9.0", "mockall 0.13.1", "prometheus", "proptest", @@ -10378,7 +10378,12 @@ name = "ic-https-outcalls-pricing" version = "0.9.0" dependencies = [ "ic-config", + "ic-logger", + "ic-metrics", "ic-types", + "ic-types-cycles", + "prometheus", + "slog", ] [[package]] diff --git a/rs/https_outcalls/client/BUILD.bazel b/rs/https_outcalls/client/BUILD.bazel index a091c40d9a78..c1825ddb1af1 100644 --- a/rs/https_outcalls/client/BUILD.bazel +++ b/rs/https_outcalls/client/BUILD.bazel @@ -25,6 +25,7 @@ rust_library( "//rs/monitoring/metrics", "//rs/types/management_canister_types", "//rs/types/types", + "//rs/utils", "@crate_index//:candid", "@crate_index//:futures", "@crate_index//:hyper-util", @@ -56,6 +57,7 @@ rust_test( "//rs/test_utilities/types", "//rs/types/management_canister_types", "//rs/types/types", + "//rs/utils", "@crate_index//:candid", "@crate_index//:futures", "@crate_index//:hyper-util", diff --git a/rs/https_outcalls/client/Cargo.toml b/rs/https_outcalls/client/Cargo.toml index bb9b7eec0758..3875900224da 100644 --- a/rs/https_outcalls/client/Cargo.toml +++ b/rs/https_outcalls/client/Cargo.toml @@ -23,6 +23,7 @@ ic-interfaces-adapter-client = { path = "../../interfaces/adapter_client" } ic-logger = { path = "../../monitoring/logger" } ic-metrics = { path = "../../monitoring/metrics" } ic-types = { path = "../../types/types" } +ic-utils = { path = "../../utils" } prometheus = { workspace = true } slog = { workspace = true } tokio = { workspace = true } diff --git a/rs/https_outcalls/client/src/client.rs b/rs/https_outcalls/client/src/client.rs index 40b1f0a5ec23..0e072520f956 100644 --- a/rs/https_outcalls/client/src/client.rs +++ b/rs/https_outcalls/client/src/client.rs @@ -15,15 +15,17 @@ use ic_logger::{ReplicaLogger, info, warn}; use ic_management_canister_types_private::{CanisterHttpResponsePayload, TransformArgs}; use ic_metrics::MetricsRegistry; use ic_types::{ - CanisterId, NumBytes, NumInstructions, + CanisterId, CountBytes, NumBytes, NumInstructions, canister_http::{ CanisterHttpHeader, CanisterHttpMethod, CanisterHttpPaymentReceipt, CanisterHttpReject, CanisterHttpRequest, CanisterHttpRequestContext, CanisterHttpResponse, - CanisterHttpResponseContent, Transform, validate_http_headers_and_body, + CanisterHttpResponseContent, MAXIMUM_CANISTER_HTTP_ERROR_MESSAGE_BYTES, Transform, + validate_http_headers_and_body, }, ingress::WasmResult, messages::{Query, QuerySource, Request}, }; +use ic_utils::str::StrEllipsize; use std::{ sync::{Arc, atomic::AtomicU64}, time::{Duration, Instant}, @@ -65,6 +67,7 @@ pub struct CanisterHttpAdapterClientImpl { rx: Receiver<(CanisterHttpResponse, CanisterHttpPaymentReceipt)>, query_service: TransformExecutionService, metrics: Metrics, + pricing_factory: PricingFactory, log: ReplicaLogger, } @@ -79,6 +82,7 @@ impl CanisterHttpAdapterClientImpl { ) -> Self { let (tx, rx) = channel(inflight_requests); let metrics = Metrics::new(&metrics_registry); + let pricing_factory = PricingFactory::new(&metrics_registry, log.clone()); Self { rt_handle, grpc_channel, @@ -86,6 +90,7 @@ impl CanisterHttpAdapterClientImpl { rx, query_service, metrics, + pricing_factory, log, } } @@ -121,6 +126,7 @@ impl NonBlockingChannel for CanisterHttpAdapterClientImpl { let mut http_adapter_client = HttpsOutcallsServiceClient::new(self.grpc_channel.clone()); let query_handler = self.query_service.clone(); let metrics = self.metrics.clone(); + let pricing_factory = self.pricing_factory.clone(); let log = self.log.clone(); // Spawn an async task that sends the canister http request to the adapter and awaits the response. @@ -131,9 +137,12 @@ impl NonBlockingChannel for CanisterHttpAdapterClientImpl { id: request_id, context: request_context, socks_proxy_addrs, + cost_schedule, + subnet_size, } = canister_http_request; - let mut budget = PricingFactory::new_tracker(&request_context); + let mut budget = + pricing_factory.new_tracker(&request_context, subnet_size, cost_schedule); let request_size = request_context.variable_parts_size(); let CanisterHttpRequestContext { @@ -149,6 +158,7 @@ impl NonBlockingChannel for CanisterHttpAdapterClientImpl { http_method: request_http_method, transform: request_transform, pricing_version: request_pricing_version, + replication: request_replication, .. } = request_context; @@ -177,7 +187,7 @@ impl NonBlockingChannel for CanisterHttpAdapterClientImpl { return; } - let payload = async { + let mut payload = async { // Execute the HTTP request and get the adapter response. let (adapter_response, downloaded_bytes, elapsed) = execute_http_request( &mut http_adapter_client, @@ -262,9 +272,42 @@ impl NonBlockingChannel for CanisterHttpAdapterClientImpl { } .await; + // Truncate an oversized reject message before pricing and gossiping + // it, so the gossip cost reflects what is actually gossiped. + if let Err(reject) = &mut payload + && reject.message.len() > MAXIMUM_CANISTER_HTTP_ERROR_MESSAGE_BYTES + { + warn!( + log, + "Pruning oversized reject message for request {}: \ + original size {}, new size {}", + request_id, + reject.message.len(), + MAXIMUM_CANISTER_HTTP_ERROR_MESSAGE_BYTES, + ); + reject.message = reject + .message + .ellipsize(MAXIMUM_CANISTER_HTTP_ERROR_MESSAGE_BYTES, 90); + } + + // Account for the cost of gossiping the final (post-transform) + // response to peers before creating the receipt. + let response_size = match &payload { + Ok(response) => response.len(), + Err(reject) => reject.count_bytes(), + }; + let payload = budget + .subtract_gossip_usage(NumBytes::from(response_size as u64)) + .map_err(|PricingError::InsufficientCycles| CanisterHttpReject { + reject_code: RejectCode::CanisterReject, + message: "Insufficient cycles".to_string(), + }) + .and(payload); + // Create the payment receipt after all processing is complete. let receipt = budget.create_payment_receipt(); + let replication = request_replication.kind(); permit.send(( CanisterHttpResponse { id: request_id, @@ -273,7 +316,11 @@ impl NonBlockingChannel for CanisterHttpAdapterClientImpl { Ok(resp) => { metrics .request_total - .with_label_values(&["success", request_http_method.as_str()]) + .with_label_values(&[ + "success", + request_http_method.as_str(), + replication.as_str(), + ]) .inc(); CanisterHttpResponseContent::Success(resp) } @@ -283,6 +330,7 @@ impl NonBlockingChannel for CanisterHttpAdapterClientImpl { .with_label_values(&[ reject.reject_code.as_str(), request_http_method.as_str(), + replication.as_str(), ]) .inc(); CanisterHttpResponseContent::Reject(reject) @@ -372,7 +420,7 @@ async fn execute_http_request( response_time: elapsed, }) .map_err(|PricingError::InsufficientCycles| CanisterHttpReject { - reject_code: RejectCode::SysFatal, + reject_code: RejectCode::CanisterReject, message: "Insufficient cycles".to_string(), })?; @@ -514,7 +562,7 @@ async fn transform_adapter_response( ); ( Err(CanisterHttpReject { - reject_code: RejectCode::SysFatal, + reject_code: RejectCode::CanisterReject, message: "Insufficient cycles".to_string(), }), instructions_used, @@ -553,6 +601,7 @@ where #[cfg(test)] mod tests { use super::*; + use ic_https_outcalls_pricing::CanisterCyclesCostSchedule; use ic_https_outcalls_service::{ HttpsOutcallRequest, HttpsOutcallResponse, HttpsOutcallResult, https_outcalls_service_server::{HttpsOutcallsService, HttpsOutcallsServiceServer}, @@ -561,7 +610,7 @@ mod tests { use ic_logger::replica_logger::no_op_logger; use ic_test_utilities_types::messages::RequestBuilder; use ic_types::{ - RegistryVersion, + NumberOfNodes, RegistryVersion, canister_http::{ MAX_CANISTER_HTTP_RESPONSE_BYTES, PricingVersion, RefundStatus, Replication, Transform, }, @@ -660,6 +709,8 @@ mod tests { registry_version: RegistryVersion::from(1), }, socks_proxy_addrs: vec![], + cost_schedule: CanisterCyclesCostSchedule::Normal, + subnet_size: NumberOfNodes::from(13), } } @@ -1128,6 +1179,109 @@ mod tests { assert_eq!(client.try_receive(), Err(TryReceiveError::Empty)); } + // Test that an oversized reject message is truncated (char-boundary-safe) + // before being returned, so that what is priced and gossiped is bounded. + #[tokio::test] + async fn test_oversized_reject_message_is_truncated() { + // Adapter mock setup. Not relevant; the transform produces the reject. + let response = HttpsOutcallResponse { + status: 200, + headers: Vec::new(), + content: Vec::new(), + }; + let mock_grpc_channel = setup_adapter_mock(Ok(create_result_from_response(response))).await; + let (svc, mut handle) = setup_system_query_mock(); + + // 300 four-byte emoji (1200 bytes) followed by a single one-byte 'x' + // (1201 bytes total). `ellipsize` keeps a prefix + "..." + suffix; the + // trailing byte makes the total length not a multiple of 4 so that the + // both suffix and prefix cut, fall *inside* an emoji. + const PREFIX_PERCENTAGE: usize = 90; + let oversized_message = format!("{}x", "😀".repeat(300)); + let oversized_len = oversized_message.len(); + assert!(oversized_len > MAXIMUM_CANISTER_HTTP_ERROR_MESSAGE_BYTES); + + // The exact byte offsets `ellipsize` cuts at. + let budget = MAXIMUM_CANISTER_HTTP_ERROR_MESSAGE_BYTES - "...".len(); + let prefix_cut = MAXIMUM_CANISTER_HTTP_ERROR_MESSAGE_BYTES * PREFIX_PERCENTAGE / 100; + let suffix_cut = oversized_len - (budget - prefix_cut); + assert!( + !oversized_message.is_char_boundary(prefix_cut), + "prefix cut at byte {prefix_cut} must fall inside a multi-byte emoji" + ); + assert!( + !oversized_message.is_char_boundary(suffix_cut), + "suffix cut at byte {suffix_cut} must fall inside a multi-byte emoji" + ); + + // The client must apply exactly this truncation before pricing the response. + let expected_message = oversized_message + .ellipsize(MAXIMUM_CANISTER_HTTP_ERROR_MESSAGE_BYTES, PREFIX_PERCENTAGE); + tokio::spawn(async move { + let (_, rsp) = handle.next_request().await.unwrap(); + rsp.send_response(Ok(( + Ok(WasmResult::Reject(oversized_message)), + current_time(), + ))); + }); + + let mut client = CanisterHttpAdapterClientImpl::new( + tokio::runtime::Handle::current(), + mock_grpc_channel, + svc, + 100, + MetricsRegistry::default(), + no_op_logger(), + ); + + assert_eq!( + client.send(build_mock_canister_http_request( + 420, + Some("transform".to_string()) + )), + Ok(()) + ); + loop { + match client.try_receive() { + Err(_) => tokio::time::sleep(Duration::from_millis(10)).await, + Ok((r, _payment_receipt)) => { + let CanisterHttpResponseContent::Reject(reject) = r.content else { + panic!("expected a reject response"); + }; + // Ellipsized exactly as the limit dictates: this pins the size + // and the ellipsize parameters. + assert_eq!( + reject.message, expected_message, + "reject message should be ellipsized to the allowed size" + ); + assert!(reject.message.len() <= MAXIMUM_CANISTER_HTTP_ERROR_MESSAGE_BYTES); + assert!(reject.message.len() < oversized_len); + + // Although both cuts fell inside an emoji (asserted above), the + // returned message is well-formed: no emoji was split. Every + // character retained on either side of the ellipsis is a whole + // emoji (plus the preserved trailing 'x' at the very end). + let (head, tail) = reject + .message + .split_once("...") + .expect("ellipsized message must contain the ellipsis"); + assert!( + !head.is_empty() && head.chars().all(|c| c == '😀'), + "prefix must consist of whole emoji, got {head:?}" + ); + assert!( + tail.strip_suffix('x') + .expect("trailing byte should be preserved") + .chars() + .all(|c| c == '😀'), + "suffix must be whole emoji followed by the trailing byte, got {tail:?}" + ); + break; + } + } + } + } + // Test client capacity. The capicity of the client is specified by the channel size. #[tokio::test] async fn test_client_at_capacity() { diff --git a/rs/https_outcalls/client/src/metrics.rs b/rs/https_outcalls/client/src/metrics.rs index fad3eae4dedd..ff82b7bcb9d9 100644 --- a/rs/https_outcalls/client/src/metrics.rs +++ b/rs/https_outcalls/client/src/metrics.rs @@ -5,6 +5,7 @@ use prometheus::{Histogram, HistogramVec, IntCounterVec}; const LABEL_STATUS_CODE: &str = "status_code"; const LABEL_HTTP_METHOD: &str = "http_method"; const LABEL_STATUS: &str = "status"; +const LABEL_REPLICATION: &str = "replication"; #[derive(Clone)] pub struct Metrics { @@ -35,7 +36,7 @@ impl Metrics { request_total: metrics_registry.int_counter_vec( "canister_http_requests_total", "Canister http request results returned to consensus.", - &[LABEL_STATUS, LABEL_HTTP_METHOD], + &[LABEL_STATUS, LABEL_HTTP_METHOD, LABEL_REPLICATION], ), } } diff --git a/rs/https_outcalls/consensus/BUILD.bazel b/rs/https_outcalls/consensus/BUILD.bazel index e0c353fd4870..dab8b0b219cb 100644 --- a/rs/https_outcalls/consensus/BUILD.bazel +++ b/rs/https_outcalls/consensus/BUILD.bazel @@ -32,7 +32,6 @@ rust_library( "//rs/types/cycles", "//rs/types/management_canister_types", "//rs/types/types", - "//rs/utils", "@crate_index//:candid", "@crate_index//:hex", "@crate_index//:prometheus", @@ -80,7 +79,6 @@ rust_test( "//rs/types/cycles", "//rs/types/management_canister_types", "//rs/types/types", - "//rs/utils", "@crate_index//:assert_matches", "@crate_index//:candid", "@crate_index//:hex", diff --git a/rs/https_outcalls/consensus/Cargo.toml b/rs/https_outcalls/consensus/Cargo.toml index a005039b5896..9a21f85adfcc 100644 --- a/rs/https_outcalls/consensus/Cargo.toml +++ b/rs/https_outcalls/consensus/Cargo.toml @@ -24,7 +24,6 @@ ic-registry-subnet-type = { path = "../../registry/subnet_type" } ic-replicated-state = { path = "../../replicated_state" } ic-types = { path = "../../types/types" } ic-types-cycles = { path = "../../types/cycles" } -ic-utils = { path = "../../utils" } prometheus = { workspace = true } rand = { workspace = true } slog = { workspace = true } diff --git a/rs/https_outcalls/consensus/src/pool_manager.rs b/rs/https_outcalls/consensus/src/pool_manager.rs index cab8d0ed7701..37b570e063f7 100644 --- a/rs/https_outcalls/consensus/src/pool_manager.rs +++ b/rs/https_outcalls/consensus/src/pool_manager.rs @@ -16,16 +16,17 @@ use ic_interfaces_registry::RegistryClient; use ic_interfaces_state_manager::StateReader; use ic_logger::*; use ic_metrics::MetricsRegistry; +use ic_protobuf::registry::subnet::v1::CanisterCyclesCostSchedule as CanisterCyclesCostScheduleProto; use ic_registry_client_helpers::api_boundary_node::ApiBoundaryNodeRegistry; use ic_registry_client_helpers::node::NodeRegistry; use ic_registry_client_helpers::subnet::SubnetRegistry; use ic_registry_subnet_type::SubnetType; use ic_replicated_state::ReplicatedState; use ic_types::{ - CountBytes, NodeId, NumBytes, ReplicaVersion, canister_http::*, crypto::Signed, - messages::CallbackId, replica_config::ReplicaConfig, + CountBytes, NodeId, NumBytes, NumberOfNodes, RegistryVersion, ReplicaVersion, canister_http::*, + crypto::Signed, messages::CallbackId, replica_config::ReplicaConfig, }; -use ic_utils::str::StrEllipsize; +use ic_types_cycles::CanisterCyclesCostSchedule; use std::{ cell::RefCell, collections::{BTreeSet, HashSet}, @@ -60,7 +61,6 @@ pub struct CanisterHttpPoolManagerImpl { log: ReplicaLogger, } -const MAXIMUM_ALLOWED_ERROR_MESSAGE_BYTES: usize = 1024; // 1KB const CANDID_OVERHEAD_RESERVE_BYTES: u64 = 1024; // 1KB // Checks that the response size is within the allowed limits. @@ -94,9 +94,9 @@ fn validate_response_size( } CanisterHttpResponseContent::Reject(reject) => { let response_size = reject.message.len(); - if response_size > MAXIMUM_ALLOWED_ERROR_MESSAGE_BYTES { + if response_size > MAXIMUM_CANISTER_HTTP_ERROR_MESSAGE_BYTES { Err(format!( - "Reject message size {response_size} exceeds the maximum allowed size of {MAXIMUM_ALLOWED_ERROR_MESSAGE_BYTES}" + "Reject message size {response_size} exceeds the maximum allowed size of {MAXIMUM_CANISTER_HTTP_ERROR_MESSAGE_BYTES}" )) } else { Ok(()) @@ -246,6 +246,26 @@ impl CanisterHttpPoolManagerImpl { .collect::>() } + /// Reads the subnet's pricing inputs (number of nodes and cycles cost + /// schedule) from the registry at `registry_version`. Returns `None` if + /// they cannot be determined. + fn pricing_inputs( + &self, + registry_version: RegistryVersion, + ) -> Option<(NumberOfNodes, CanisterCyclesCostSchedule)> { + let record = self + .registry_client + .get_subnet_record(self.replica_config.subnet_id, registry_version) + .ok() + .flatten()?; + let subnet_size = u32::try_from(record.membership.len()).ok()?; + let cost_schedule = + CanisterCyclesCostScheduleProto::try_from(record.canister_cycles_cost_schedule) + .ok() + .map(CanisterCyclesCostSchedule::from)?; + Some((NumberOfNodes::from(subnet_size), cost_schedule)) + } + /// Returns whether `node_id` belongs to the committee responsible for the /// given request, evaluated at the registry version pinned in the request /// context. @@ -326,6 +346,20 @@ impl CanisterHttpPoolManagerImpl { } if !request_ids_already_made.contains(id) { + let Some((subnet_size, cost_schedule)) = + self.pricing_inputs(context.registry_version) + else { + warn!( + every_n_seconds => 10, + self.log, + "Skipping canister http request {} because the subnet size or cost \ + schedule could not be determined at registry version {}", + id, + context.registry_version + ); + continue; + }; + if let Err(err) = self .http_adapter_shim .lock() @@ -334,6 +368,8 @@ impl CanisterHttpPoolManagerImpl { id: *id, context: context.clone(), socks_proxy_addrs: socks_proxy_addrs.clone(), + cost_schedule, + subnet_size, }) { warn!( @@ -366,7 +402,7 @@ impl CanisterHttpPoolManagerImpl { loop { match self.http_adapter_shim.lock().unwrap().try_receive() { Err(TryReceiveError::Empty) => break, - Ok((mut response, payment_receipt)) => { + Ok((response, payment_receipt)) => { // Drop the response if its context is no longer present in the replicated state // (e.g. the request has timed out or has already been answered by enough other nodes). let Some(context) = active_contexts.get(&response.id) else { @@ -380,28 +416,6 @@ impl CanisterHttpPoolManagerImpl { continue; }; - // Truncate the reject message if it's too long. - // - // The "happy path" response is organically bounded by max_response_bytes, however we need to set a - // limit for the error message as well. - // - // The current limit is 1KB, which should be reasonable for an error message. - if let CanisterHttpResponseContent::Reject(reject) = &mut response.content - && reject.message.len() > MAXIMUM_ALLOWED_ERROR_MESSAGE_BYTES - { - let original_len = reject.message.len(); - warn!( - self.log, - "Pruning oversized reject message for request ID {}. Original size: {}, New size: {}", - response.id, - original_len, - MAXIMUM_ALLOWED_ERROR_MESSAGE_BYTES - ); - reject.message = reject - .message - .ellipsize(MAXIMUM_ALLOWED_ERROR_MESSAGE_BYTES, 90); - } - let receipt_share = CanisterHttpResponseReceipt { metadata: CanisterHttpResponseMetadata { id: response.id, @@ -1699,7 +1713,7 @@ pub mod test { ); // 3. ARTIFACT: Create a Reject response. Its message size is valid - // (i.e., less than MAXIMUM_ALLOWED_ERROR_MESSAGE_BYTES), so it should pass + // (i.e., less than MAXIMUM_CANISTER_HTTP_ERROR_MESSAGE_BYTES), so it should pass // validation despite the context's zero-byte limit for success responses. let mut canister_http_pool = CanisterHttpPoolImpl::new(MetricsRegistry::new(), no_op_logger()); @@ -1754,262 +1768,6 @@ pub mod test { }); } - #[test] - fn test_oversized_reject_message_is_pruned_not_invalidated() { - ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| { - with_test_replica_logger(|log| { - // 1. SETUP: Standard dependencies. - let Dependencies { - pool, - replica_config, - crypto, - state_manager, - registry, - .. - } = dependencies(pool_config.clone(), 4); - - let callback_id = CallbackId::from(5); - // Delegate the request to this node so it is the authorized signer - // and creates a share for the injected response. - let delegated_node_id = replica_config.node_id; - - // 2. CONTEXT: Set up a NonReplicated request context in the state manager. - // This ensures we test the gossiping code path. - let request_context = test_request_context( - Replication::NonReplicated(delegated_node_id), - PricingVersion::Legacy, - None, - ); - state_manager - .get_mut() - .expect_get_latest_state() - .return_const(Labeled::new( - Height::from(1), - Arc::new(state_with_pending_http_calls(BTreeMap::from([( - callback_id, - request_context, - )]))), - )); - - // 3. OVERSIZED RESPONSE: Define an error message that is intentionally too large. - let oversized_len = MAXIMUM_ALLOWED_ERROR_MESSAGE_BYTES + 100; - let max_len = MAXIMUM_ALLOWED_ERROR_MESSAGE_BYTES; - let oversized_message = "a".repeat(oversized_len); - - let oversized_response = CanisterHttpResponse { - id: callback_id, - content: CanisterHttpResponseContent::Reject(CanisterHttpReject { - reject_code: RejectCode::SysFatal, - message: oversized_message, - }), - ..empty_canister_http_response(callback_id.get()) - }; - - // 4. MOCK ADAPTER: Mock the adapter to return the oversized response once. - let mut shim_mock = MockNonBlockingChannel::::new(); - let mut sequence = Sequence::new(); - shim_mock - .expect_try_receive() - .times(1) - .returning(move || { - Ok(( - oversized_response.clone(), - CanisterHttpPaymentReceipt::default(), - )) - }) - .in_sequence(&mut sequence); - shim_mock - .expect_try_receive() - .times(1) - .returning(|| Err(TryReceiveError::Empty)) - .in_sequence(&mut sequence); - - let shim: Arc> = - Arc::new(Mutex::new(Box::new(shim_mock))); - - let pool_manager = CanisterHttpPoolManagerImpl::new( - state_manager, - shim, - crypto, - pool.get_cache(), - replica_config, - SubnetType::Application, - Arc::clone(®istry) as Arc<_>, - MetricsRegistry::new(), - log, - ); - - // 5. ACTION: Call the function to generate shares from responses. - let change_set = pool_manager.create_shares_from_responses(); - - // 6. ASSERTIONS: - assert_eq!( - change_set.len(), - 1, - "A change action should have been created" - ); - - // Check that the action contains the *pruned* response and a share with the correct hash. - assert_matches!( - &change_set[0], - CanisterHttpChangeAction::AddToValidatedAndGossipResponse(share, response) => { - // Assert that the response message was indeed truncated. - let pruned_message_len = if let CanisterHttpResponseContent::Reject(r) = &response.content { - r.message.len() - } else { - panic!("Test failed: Expected a Reject response content"); - }; - assert_eq!( - pruned_message_len, max_len, - "The reject message should have been truncated to the maximum allowed length" - ); - - // Assert that the hash in the share matches the hash of the *truncated* response. - let expected_hash = ic_types::crypto::crypto_hash(response); - assert_eq!( - share.content.content_hash(), &expected_hash, - "The share's content hash must match the pruned response" - ); - } - ); - }) - }); - } - - #[test] - fn test_oversized_reject_with_multibyte_char_is_pruned_safely() { - // This test ensures that when pruning an oversized reject message, - // the logic correctly handles multi-byte UTF-8 characters that - // straddle the byte limit, preventing a panic that a simple `truncate()` would cause. - ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| { - with_test_replica_logger(|log| { - // 1. SETUP: Standard dependencies. - let Dependencies { - pool, - replica_config, - crypto, - state_manager, - registry, - .. - } = dependencies(pool_config.clone(), 4); - - let callback_id = CallbackId::from(0); // Use ID that will pass validation filter - // Delegate the request to this node so it is the authorized signer - // and creates a share for the injected response. - let delegated_node_id = replica_config.node_id; - - // 2. CONTEXT: Set up a NonReplicated request context. - let request_context = test_request_context( - Replication::NonReplicated(delegated_node_id), - PricingVersion::Legacy, - None, - ); - state_manager - .get_mut() - .expect_get_latest_state() - .return_const(Labeled::new( - Height::from(1), - Arc::new(state_with_pending_http_calls(BTreeMap::from([( - callback_id, - request_context, - )]))), - )); - - // 3. MALICIOUS RESPONSE: Construct a string where a 4-byte emoji ('👍') - // starts 2 bytes before the limit, causing it to cross the boundary. - let max_len = MAXIMUM_ALLOWED_ERROR_MESSAGE_BYTES; - let padding = "a".repeat(max_len - 2); - let emoji = "👍"; // A 4-byte character - let oversized_message = format!("{}{}", padding, emoji); - - // Verify the setup: the message is oversized and the emoji crosses the boundary. - assert!(oversized_message.len() > max_len); - assert_eq!(padding.len(), max_len - 2); - - let oversized_response = CanisterHttpResponse { - id: callback_id, - content: CanisterHttpResponseContent::Reject(CanisterHttpReject { - reject_code: RejectCode::SysFatal, - message: oversized_message, - }), - ..empty_canister_http_response(callback_id.get()) - }; - - // 4. MOCK ADAPTER: Mock the adapter to return this specific response. - let mut shim_mock = MockNonBlockingChannel::::new(); - shim_mock - .expect_try_receive() - .times(1) - .return_once(move || { - Ok(( - oversized_response.clone(), - CanisterHttpPaymentReceipt::default(), - )) - }); - shim_mock - .expect_try_receive() - .return_const(Err(TryReceiveError::Empty)); - - let shim: Arc> = - Arc::new(Mutex::new(Box::new(shim_mock))); - - let pool_manager = CanisterHttpPoolManagerImpl::new( - state_manager, - shim, - crypto, - pool.get_cache(), - replica_config, - SubnetType::Application, - Arc::clone(®istry) as Arc<_>, - MetricsRegistry::new(), - log, - ); - - // 5. ACTION: Call the function. The test will fail with a panic if - // the pruning logic is incorrect (i.e., not UTF-8 aware). - let change_set = pool_manager.create_shares_from_responses(); - - // 6. ASSERTIONS: - assert_eq!( - change_set.len(), - 1, - "A change action should have been created" - ); - - assert_matches!( - &change_set[0], - CanisterHttpChangeAction::AddToValidatedAndGossipResponse(share, response) => { - let pruned_message = if let CanisterHttpResponseContent::Reject(r) = &response.content { - &r.message - } else { - panic!("Test failed: Expected a Reject response content"); - }; - - // Assert that the pruned message is now within the byte limit. - assert!( - pruned_message.len() <= max_len, - "The pruned message (len: {}) should not exceed the max length ({})", - pruned_message.len(), max_len - ); - - // Assert the emoji is still present because ellipsize preserves the end of the string. - assert!( - pruned_message.contains(emoji), - "Pruned message should still contain the emoji from the suffix" - ); - - // Assert that the hash in the share matches the hash of the now-pruned response. - let expected_hash = ic_types::crypto::crypto_hash(response); - assert_eq!( - share.content.content_hash(), &expected_hash, - "The share's content hash must match the pruned response" - ); - } - ); - }) - }); - } - #[test] fn test_dishonest_oversized_reject_is_invalidated() { ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| { @@ -2046,7 +1804,7 @@ pub mod test { )); // 3. DISHONEST ARTIFACT: - let oversized_len = MAXIMUM_ALLOWED_ERROR_MESSAGE_BYTES + 1; + let oversized_len = MAXIMUM_CANISTER_HTTP_ERROR_MESSAGE_BYTES + 1; let dishonest_response = CanisterHttpResponse { id: callback_id, content: CanisterHttpResponseContent::Reject(CanisterHttpReject { @@ -2109,7 +1867,7 @@ pub mod test { let expected_error = format!( "Http Response for request ID {} is too large: Reject message size {} exceeds the maximum allowed size of {}", - callback_id, oversized_len, MAXIMUM_ALLOWED_ERROR_MESSAGE_BYTES + callback_id, oversized_len, MAXIMUM_CANISTER_HTTP_ERROR_MESSAGE_BYTES ); assert_matches!( &change_set[0], @@ -2181,7 +1939,7 @@ pub mod test { let reject_message = "This error message is definitely longer than 10 bytes.".to_string(); assert!(reject_message.len() as u64 > LOW_MAX_RESPONSE_BYTES); - assert!(reject_message.len() <= MAXIMUM_ALLOWED_ERROR_MESSAGE_BYTES); + assert!(reject_message.len() <= MAXIMUM_CANISTER_HTTP_ERROR_MESSAGE_BYTES); let reject_content = CanisterHttpReject { reject_code: RejectCode::SysFatal, @@ -2640,6 +2398,8 @@ pub mod test { id: CallbackId::from(7), context: request.clone(), socks_proxy_addrs: vec![], + cost_schedule: CanisterCyclesCostSchedule::Normal, + subnet_size: NumberOfNodes::from(4), })) .times(1) .return_const(Ok(())); diff --git a/rs/https_outcalls/pricing/BUILD.bazel b/rs/https_outcalls/pricing/BUILD.bazel index bb6fa3989628..2c948a5854c5 100644 --- a/rs/https_outcalls/pricing/BUILD.bazel +++ b/rs/https_outcalls/pricing/BUILD.bazel @@ -8,8 +8,14 @@ rust_library( crate_name = "ic_https_outcalls_pricing", version = "0.1.0", deps = [ + # Keep sorted. "//rs/config", + "//rs/monitoring/logger", + "//rs/monitoring/metrics", + "//rs/types/cycles", "//rs/types/types", + "@crate_index//:prometheus", + "@crate_index//:slog", ], ) diff --git a/rs/https_outcalls/pricing/Cargo.toml b/rs/https_outcalls/pricing/Cargo.toml index 1773aa69401f..c3eeb858fe1e 100644 --- a/rs/https_outcalls/pricing/Cargo.toml +++ b/rs/https_outcalls/pricing/Cargo.toml @@ -8,4 +8,9 @@ documentation.workspace = true [dependencies] ic-config = { path = "../../config" } -ic-types = { path = "../../types/types" } \ No newline at end of file +ic-logger = { path = "../../monitoring/logger" } +ic-metrics = { path = "../../monitoring/metrics" } +ic-types = { path = "../../types/types" } +ic-types-cycles = { path = "../../types/cycles" } +prometheus = { workspace = true } +slog = { workspace = true } diff --git a/rs/https_outcalls/pricing/src/dark_launch.rs b/rs/https_outcalls/pricing/src/dark_launch.rs new file mode 100644 index 000000000000..cab05b73617a --- /dev/null +++ b/rs/https_outcalls/pricing/src/dark_launch.rs @@ -0,0 +1,305 @@ +use ic_logger::{ReplicaLogger, warn}; +use ic_types::{ + CanisterId, NumBytes, NumInstructions, + canister_http::{CanisterHttpPaymentReceipt, ReplicationKind}, +}; + +use crate::{AdapterLimits, BudgetTracker, NetworkUsage, PricingError, metrics::PricingMetrics}; + +/// A [`BudgetTracker`] that runs two child trackers side by side: a `real` +/// tracker whose results are the only ones returned (and therefore the only +/// ones that affect observable behaviour), and a `shadow` tracker whose results +/// are merely compared against the real one. +/// +/// A counter metric is increased whenever the shadow tracker computes a different +/// result than the real tracker. This could happen if the shadow tracker returns +/// a pricing error where the real tracker succeeded, meaning it ran out of cycles. +pub struct DarkLaunchTracker { + real: Box, + shadow: Box, + canister_id: CanisterId, + replication: ReplicationKind, + metrics: PricingMetrics, + log: ReplicaLogger, + /// Whether an incompatibility has already been recorded for this request. + /// Ensures we count and log at most once per request. + error_reported: bool, +} + +impl DarkLaunchTracker { + pub fn new( + real: Box, + shadow: Box, + canister_id: CanisterId, + replication: ReplicationKind, + metrics: PricingMetrics, + log: ReplicaLogger, + ) -> Self { + Self { + real, + shadow, + canister_id, + replication, + metrics, + log, + error_reported: false, + } + } + + /// Compares the results of the real and shadow trackers for a given + /// accounting `step` and increment the shadow_incompatible_total metric + /// if they differ. + fn compare( + &mut self, + step: &str, + real: &Result<(), PricingError>, + shadow: &Result<(), PricingError>, + ) { + if real.is_err() || shadow.is_ok() { + return; + } + if self.error_reported { + return; + } + self.error_reported = true; + self.metrics + .shadow_incompatible_total + .with_label_values(&[step, self.replication.as_str()]) + .inc(); + warn!( + self.log, + "Canister http request would not be compatible under shadow pricing: \ + canister_id {}, step {}, replication {}, real_result {:?}, shadow_result {:?}", + self.canister_id, + step, + self.replication.as_str(), + real, + shadow, + ); + } +} + +impl BudgetTracker for DarkLaunchTracker { + fn get_adapter_limits(&self) -> AdapterLimits { + self.real.get_adapter_limits() + } + + fn subtract_network_usage(&mut self, network_usage: NetworkUsage) -> Result<(), PricingError> { + let real = self.real.subtract_network_usage(network_usage); + let shadow = self.shadow.subtract_network_usage(network_usage); + self.compare("network_usage", &real, &shadow); + real + } + + fn get_transform_limit(&self) -> NumInstructions { + self.real.get_transform_limit() + } + + fn subtract_transform_usage(&mut self, usage: NumInstructions) -> Result<(), PricingError> { + let real = self.real.subtract_transform_usage(usage); + let shadow = self.shadow.subtract_transform_usage(usage); + self.compare("transform_usage", &real, &shadow); + real + } + + fn subtract_gossip_usage( + &mut self, + transformed_response_size: NumBytes, + ) -> Result<(), PricingError> { + let real = self.real.subtract_gossip_usage(transformed_response_size); + let shadow = self.shadow.subtract_gossip_usage(transformed_response_size); + self.compare("gossip_usage", &real, &shadow); + real + } + + fn create_payment_receipt(&self) -> CanisterHttpPaymentReceipt { + self.real.create_payment_receipt() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use ic_logger::no_op_logger; + use ic_metrics::MetricsRegistry; + use std::time::Duration; + + /// A [`BudgetTracker`] whose accounting steps return preconfigured results. + struct FakeTracker { + network: Result<(), PricingError>, + transform: Result<(), PricingError>, + transformed: Result<(), PricingError>, + } + + impl FakeTracker { + fn ok() -> Self { + Self { + network: Ok(()), + transform: Ok(()), + transformed: Ok(()), + } + } + } + + impl BudgetTracker for FakeTracker { + fn get_adapter_limits(&self) -> AdapterLimits { + AdapterLimits { + max_response_size: NumBytes::from(0), + max_response_time: Duration::ZERO, + } + } + fn subtract_network_usage(&mut self, _: NetworkUsage) -> Result<(), PricingError> { + self.network + } + fn get_transform_limit(&self) -> NumInstructions { + NumInstructions::from(0) + } + fn subtract_transform_usage(&mut self, _: NumInstructions) -> Result<(), PricingError> { + self.transform + } + fn subtract_gossip_usage(&mut self, _: NumBytes) -> Result<(), PricingError> { + self.transformed + } + fn create_payment_receipt(&self) -> CanisterHttpPaymentReceipt { + CanisterHttpPaymentReceipt::default() + } + } + + fn dark_launch( + real: FakeTracker, + shadow: FakeTracker, + replication: ReplicationKind, + metrics: PricingMetrics, + ) -> DarkLaunchTracker { + DarkLaunchTracker::new( + Box::new(real), + Box::new(shadow), + CanisterId::from_u64(7), + replication, + metrics, + no_op_logger(), + ) + } + + fn network_usage() -> NetworkUsage { + NetworkUsage { + response_size: NumBytes::from(0), + response_time: Duration::ZERO, + } + } + + fn incompatible_count(metrics: &PricingMetrics) -> u64 { + let mut total = 0; + for step in ["network_usage", "transform_usage", "gossip_usage"] { + for replication in ["fully_replicated", "flexible", "non_replicated"] { + total += metrics + .shadow_incompatible_total + .with_label_values(&[step, replication]) + .get(); + } + } + total + } + + #[test] + fn returns_real_result_and_increments_counter() { + let metrics = PricingMetrics::new(&MetricsRegistry::new()); + let shadow = FakeTracker { + network: Err(PricingError::InsufficientCycles), + ..FakeTracker::ok() + }; + let mut tracker = dark_launch( + FakeTracker::ok(), + shadow, + ReplicationKind::FullyReplicated, + metrics.clone(), + ); + + // The real (always-Ok) result is returned even though the shadow fails. + assert_eq!(tracker.subtract_network_usage(network_usage()), Ok(())); + assert_eq!( + metrics + .shadow_incompatible_total + .with_label_values(&["network_usage", "fully_replicated"]) + .get(), + 1 + ); + } + + #[test] + fn counts_incompatible_requests_at_most_once_per_request() { + let metrics = PricingMetrics::new(&MetricsRegistry::new()); + let shadow = FakeTracker { + network: Err(PricingError::InsufficientCycles), + transform: Err(PricingError::InsufficientCycles), + transformed: Err(PricingError::InsufficientCycles), + }; + let mut tracker = dark_launch( + FakeTracker::ok(), + shadow, + ReplicationKind::Flexible, + metrics.clone(), + ); + + assert_eq!(tracker.subtract_network_usage(network_usage()), Ok(())); + assert_eq!( + tracker.subtract_transform_usage(NumInstructions::from(0)), + Ok(()) + ); + assert_eq!(tracker.subtract_gossip_usage(NumBytes::from(0)), Ok(())); + + // Only the first occurance is recorded for the request. + assert_eq!(incompatible_count(&metrics), 1); + } + + #[test] + fn do_not_increase_counter_when_results_agree() { + let metrics = PricingMetrics::new(&MetricsRegistry::new()); + let mut tracker = dark_launch( + FakeTracker::ok(), + FakeTracker::ok(), + ReplicationKind::FullyReplicated, + metrics.clone(), + ); + + assert_eq!(tracker.subtract_network_usage(network_usage()), Ok(())); + assert_eq!( + tracker.subtract_transform_usage(NumInstructions::from(0)), + Ok(()) + ); + assert_eq!(incompatible_count(&metrics), 0); + } + + #[test] + fn labels_incompatibility_by_replication_type() { + let metrics = PricingMetrics::new(&MetricsRegistry::new()); + let shadow = FakeTracker { + network: Err(PricingError::InsufficientCycles), + ..FakeTracker::ok() + }; + let mut tracker = dark_launch( + FakeTracker::ok(), + shadow, + ReplicationKind::NonReplicated, + metrics.clone(), + ); + + assert_eq!(tracker.subtract_network_usage(network_usage()), Ok(())); + + // The incompatibility is attributed to the non_replicated label. + assert_eq!( + metrics + .shadow_incompatible_total + .with_label_values(&["network_usage", "non_replicated"]) + .get(), + 1 + ); + assert_eq!( + metrics + .shadow_incompatible_total + .with_label_values(&["network_usage", "fully_replicated"]) + .get(), + 0 + ); + } +} diff --git a/rs/https_outcalls/pricing/src/legacy.rs b/rs/https_outcalls/pricing/src/legacy.rs index eb272bc8a37b..3cde14d3284d 100644 --- a/rs/https_outcalls/pricing/src/legacy.rs +++ b/rs/https_outcalls/pricing/src/legacy.rs @@ -1,12 +1,10 @@ -use std::time::Duration; - use ic_config::subnet_config::MAX_INSTRUCTIONS_PER_QUERY_MESSAGE; use ic_types::{ NumBytes, NumInstructions, canister_http::{CanisterHttpPaymentReceipt, MAX_CANISTER_HTTP_RESPONSE_BYTES}, }; -use crate::{AdapterLimits, BudgetTracker, NetworkUsage, PricingError}; +use crate::{AdapterLimits, BudgetTracker, MAX_RESPONSE_TIME, NetworkUsage, PricingError}; pub struct LegacyTracker { max_response_size: NumBytes, @@ -25,9 +23,7 @@ impl BudgetTracker for LegacyTracker { fn get_adapter_limits(&self) -> AdapterLimits { AdapterLimits { max_response_size: self.max_response_size, - // Note: there is already a timeout limit on the server itself (30 seconds, see DEFAULT_HTTP_REQUEST_TIMEOUT_SECS). - // Setting higher than that just to be safe. - max_response_time: Duration::from_secs(60), + max_response_time: MAX_RESPONSE_TIME, } } @@ -45,6 +41,13 @@ impl BudgetTracker for LegacyTracker { Ok(()) } + fn subtract_gossip_usage( + &mut self, + _transformed_response_size: NumBytes, + ) -> Result<(), PricingError> { + Ok(()) + } + fn create_payment_receipt(&self) -> CanisterHttpPaymentReceipt { // Legacy pricing does not perform cycles accounting, so no cycles // are ever refunded. diff --git a/rs/https_outcalls/pricing/src/lib.rs b/rs/https_outcalls/pricing/src/lib.rs index 827459a9bcdd..c1b5f555f826 100644 --- a/rs/https_outcalls/pricing/src/lib.rs +++ b/rs/https_outcalls/pricing/src/lib.rs @@ -1,17 +1,27 @@ +mod dark_launch; mod legacy; +mod metrics; +mod payg; use std::time::Duration; +use ic_logger::ReplicaLogger; +use ic_metrics::MetricsRegistry; use ic_types::{ - NumBytes, NumInstructions, - canister_http::{CanisterHttpPaymentReceipt, CanisterHttpRequestContext}, + NumBytes, NumInstructions, NumberOfNodes, + canister_http::{CanisterHttpPaymentReceipt, CanisterHttpRequestContext, PricingVersion}, }; +pub use ic_types_cycles::CanisterCyclesCostSchedule; + +use dark_launch::DarkLaunchTracker; use legacy::LegacyTracker; +use metrics::PricingMetrics; +use payg::PayAsYouGoTracker; pub trait BudgetTracker: Send { /// Returns the maximum network resources the Adapter is allowed to consume. fn get_adapter_limits(&self) -> AdapterLimits; - /// Deducts the actual network resources consumed. + /// Deducts the cost of the network resources consumed by the request. /// /// # Invariants /// - This method returns `Ok(())` if `network_usage <= get_adapter_limits()`. @@ -21,17 +31,33 @@ pub trait BudgetTracker: Send { fn subtract_network_usage(&mut self, network_usage: NetworkUsage) -> Result<(), PricingError>; /// Returns the maximum instructions allowed for the transformation function. fn get_transform_limit(&self) -> NumInstructions; - /// Deducts the actual instructions consumed by the transformation. + /// Deducts the cost of the instructions consumed by the transformation. /// /// # Invariants /// - This method returns `Ok(())` if and only if `usage <= get_transform_limit()`. fn subtract_transform_usage(&mut self, usage: NumInstructions) -> Result<(), PricingError>; + /// Deducts the cost of the final (post-transform) response that this replica + /// produced and that will be gossiped to peers. This cost does not apply to fully-replicated + /// requests, which doesn't gossip responses. + /// + /// This is the last accounting step and is invoked once the size of the + /// response is known. + fn subtract_gossip_usage( + &mut self, + transformed_response_size: NumBytes, + ) -> Result<(), PricingError>; /// Produces the per-replica payment receipt that summarizes the cycles /// accounting outcome of the outcall, given the resources consumed so /// far via the `subtract_*` methods. fn create_payment_receipt(&self) -> CanisterHttpPaymentReceipt; } +/// The maximum duration the adapter is allowed to take to fully receive a +/// response, as measured by the client. The server already enforces a 30s +/// timeout (see `DEFAULT_HTTP_REQUEST_TIMEOUT_SECS`), so this is a safety margin +/// above it. +pub(crate) const MAX_RESPONSE_TIME: Duration = Duration::from_secs(60); + pub struct AdapterLimits { /// The maximum size of the HTTP response, including the headers and the body. pub max_response_size: NumBytes, @@ -39,6 +65,7 @@ pub struct AdapterLimits { pub max_response_time: Duration, } +#[derive(Clone, Copy)] pub struct NetworkUsage { /// The size of the HTTP response, including the headers and the body. pub response_size: NumBytes, @@ -46,16 +73,43 @@ pub struct NetworkUsage { pub response_time: Duration, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum PricingError { InsufficientCycles, } -pub struct PricingFactory; +#[derive(Clone)] +pub struct PricingFactory { + metrics: PricingMetrics, + log: ReplicaLogger, +} impl PricingFactory { - pub fn new_tracker(context: &CanisterHttpRequestContext) -> Box { - // TODO(IC-1937): This should take into account context.pricing_version and a replica config. - // Currently, we only support the legacy pricing version. - Box::new(LegacyTracker::new(context.max_response_bytes)) + pub fn new(metrics_registry: &MetricsRegistry, log: ReplicaLogger) -> Self { + Self { + metrics: PricingMetrics::new(metrics_registry), + log, + } + } + + pub fn new_tracker( + &self, + context: &CanisterHttpRequestContext, + subnet_size: NumberOfNodes, + cost_schedule: CanisterCyclesCostSchedule, + ) -> Box { + match context.pricing_version { + PricingVersion::Legacy => Box::new(DarkLaunchTracker::new( + Box::new(LegacyTracker::new(context.max_response_bytes)), + Box::new(PayAsYouGoTracker::new(context, subnet_size, cost_schedule)), + context.request.sender, + context.replication.kind(), + self.metrics.clone(), + self.log.clone(), + )), + PricingVersion::PayAsYouGo => { + Box::new(PayAsYouGoTracker::new(context, subnet_size, cost_schedule)) + } + } } } diff --git a/rs/https_outcalls/pricing/src/metrics.rs b/rs/https_outcalls/pricing/src/metrics.rs new file mode 100644 index 000000000000..f4565d03f843 --- /dev/null +++ b/rs/https_outcalls/pricing/src/metrics.rs @@ -0,0 +1,30 @@ +use ic_metrics::MetricsRegistry; +use prometheus::IntCounterVec; + +/// Label identifying the accounting step at which the shadow tracker ran out of cycles +/// before the real one. +pub const LABEL_STEP: &str = "step"; + +/// Label identifying the replication kind of the request. +pub const LABEL_REPLICATION: &str = "replication"; + +#[derive(Clone)] +pub struct PricingMetrics { + /// Number of requests for which the shadow tracker ran out of cycles before the + /// real tracker, by the accounting step at which the incompatibility was first + /// observed and by replication kind. + pub shadow_incompatible_total: IntCounterVec, +} + +impl PricingMetrics { + pub fn new(metrics_registry: &MetricsRegistry) -> Self { + Self { + shadow_incompatible_total: metrics_registry.int_counter_vec( + "canister_http_pricing_shadow_incompatible_total", + "Canister http requests that attached enough cycles to be compatible with the + real tracker, but not for the shadow tracker.", + &[LABEL_STEP, LABEL_REPLICATION], + ), + } + } +} diff --git a/rs/https_outcalls/pricing/src/payg.rs b/rs/https_outcalls/pricing/src/payg.rs new file mode 100644 index 000000000000..b77069604d8d --- /dev/null +++ b/rs/https_outcalls/pricing/src/payg.rs @@ -0,0 +1,344 @@ +use ic_config::subnet_config::MAX_INSTRUCTIONS_PER_QUERY_MESSAGE; +use ic_types::{ + NumBytes, NumInstructions, NumberOfNodes, + canister_http::{ + CanisterHttpPaymentReceipt, CanisterHttpRequestContext, MAX_CANISTER_HTTP_RESPONSE_BYTES, + Replication, + }, +}; +use ic_types_cycles::{CanisterCyclesCostSchedule, Cycles}; + +use crate::{AdapterLimits, BudgetTracker, MAX_RESPONSE_TIME, NetworkUsage, PricingError}; + +// Per-replica fee constants. +// +// A request's cost is split into three parts: +// 1. the base cost, subtracted up-front when the request context is created +// (and therefore reflected in `per_replica_allowance`); +// 2. the per-replica cost, accounted for here as-you-go; +// 3. the consensus cost, computed from the aggregated response in the block +// payload. +// +// This tracker implements the per-replica part. The formula differs +// between fully-replicated and non-replicated/flexible outcalls: +// +// Fully-replicated per replica: +// 50 * downloaded_bytes_i + 300 * request_ms_i + transform_instructions_i / 13 +// +// Non-replicated/Flexible per replica: +// 50 * downloaded_bytes_i + 300 * request_ms_i +// + 50 * transformed_response_bytes_i * N + transform_instructions_i / 13 +const PER_DOWNLOADED_BYTE_FEE: u128 = 50; +const PER_RESPONSE_MS_FEE: u128 = 300; +/// HTTP outcalls are priced consistently against a reference subnet size of 13. +const TRANSFORM_INSTRUCTION_DIVISOR: u128 = 13; +const FLEXIBLE_PER_TRANSFORMED_BYTE_NODE_FEE: u128 = 50; + +pub struct PayAsYouGoTracker { + /// Number of nodes (`N`) on the subnet. + subnet_size: NumberOfNodes, + /// Whether responses to this outcalls are gossiped (only flexible and non-replicated). + is_gossiping: bool, + /// Whether the subnet uses a free cost schedule. When `true` the tracker + /// charges nothing and refunds the full allowance. + is_free: bool, + /// The cycles budget available to this replica (already net of the base + /// cost, which was subtracted when the context was created). + allowance: u128, + /// The maximum size of the HTTP response, including headers and body. + max_response_size: NumBytes, + /// The cycles charged so far against `allowance`. + spent: u128, +} + +impl PayAsYouGoTracker { + pub fn new( + context: &CanisterHttpRequestContext, + subnet_size: NumberOfNodes, + cost_schedule: CanisterCyclesCostSchedule, + ) -> Self { + Self { + subnet_size, + is_gossiping: match context.replication { + // Non-replicated outcalls gossip the response, so they are charged + // the same way as flexible outcalls. + Replication::Flexible { .. } | Replication::NonReplicated(_) => true, + Replication::FullyReplicated => false, + }, + is_free: match cost_schedule { + CanisterCyclesCostSchedule::Free => true, + CanisterCyclesCostSchedule::Normal => false, + }, + allowance: context.refund_status.per_replica_allowance.get(), + max_response_size: context + .max_response_bytes + .unwrap_or(NumBytes::from(MAX_CANISTER_HTTP_RESPONSE_BYTES)), + spent: 0, + } + } + + /// Charges `amount` against the budget. Returns an error if the total spent + /// now exceeds the available allowance. + fn charge(&mut self, amount: u128) -> Result<(), PricingError> { + // A free cost schedule means the subnet charges nothing for resources. + if self.is_free { + return Ok(()); + } + self.spent = self.spent.saturating_add(amount); + if self.spent > self.allowance { + Err(PricingError::InsufficientCycles) + } else { + Ok(()) + } + } +} + +impl BudgetTracker for PayAsYouGoTracker { + fn get_adapter_limits(&self) -> AdapterLimits { + // TODO: Adjust limits based on remaining budget. + AdapterLimits { + max_response_size: self.max_response_size, + max_response_time: MAX_RESPONSE_TIME, + } + } + + fn subtract_network_usage(&mut self, network_usage: NetworkUsage) -> Result<(), PricingError> { + let NetworkUsage { + response_size, + response_time, + } = network_usage; + let cost = PER_DOWNLOADED_BYTE_FEE + .saturating_mul(response_size.get() as u128) + .saturating_add(PER_RESPONSE_MS_FEE.saturating_mul(response_time.as_millis())); + self.charge(cost) + } + + fn get_transform_limit(&self) -> NumInstructions { + // TODO: Adjust limits based on remaining budget. + MAX_INSTRUCTIONS_PER_QUERY_MESSAGE + } + + fn subtract_transform_usage(&mut self, usage: NumInstructions) -> Result<(), PricingError> { + let cost = (usage.get() as u128) / TRANSFORM_INSTRUCTION_DIVISOR; + self.charge(cost) + } + + fn subtract_gossip_usage( + &mut self, + transformed_response_size: NumBytes, + ) -> Result<(), PricingError> { + // For fully replicated outcalls the gossip term is a + // consensus cost (ignored here). For flexible outcalls each + // replica is charged 50 * transformed_response_bytes_i * N. + if !self.is_gossiping { + return Ok(()); + } + let cost = FLEXIBLE_PER_TRANSFORMED_BYTE_NODE_FEE + .saturating_mul(transformed_response_size.get() as u128) + .saturating_mul(self.subnet_size.get() as u128); + self.charge(cost) + } + + fn create_payment_receipt(&self) -> CanisterHttpPaymentReceipt { + CanisterHttpPaymentReceipt { + refund: Cycles::new(self.allowance.saturating_sub(self.spent)), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use ic_types::{ + CanisterId, NodeId, PrincipalId, RegistryVersion, + canister_http::{CanisterHttpMethod, PricingVersion, RefundStatus}, + messages::{CallbackId, NO_DEADLINE, Request}, + time::UNIX_EPOCH, + }; + use std::collections::BTreeSet; + use std::time::Duration; + + fn context( + replication: Replication, + per_replica_allowance: u128, + ) -> CanisterHttpRequestContext { + CanisterHttpRequestContext { + request: Request { + receiver: CanisterId::from_u64(1), + sender: CanisterId::from_u64(1), + sender_reply_callback: CallbackId::from(1), + payment: Cycles::zero(), + method_name: String::new(), + method_payload: Vec::new(), + metadata: Default::default(), + deadline: NO_DEADLINE, + }, + url: String::new(), + max_response_bytes: None, + headers: vec![], + body: None, + http_method: CanisterHttpMethod::GET, + transform: None, + time: UNIX_EPOCH, + replication, + pricing_version: PricingVersion::Legacy, + refund_status: RefundStatus { + refundable_cycles: Cycles::new(per_replica_allowance), + per_replica_allowance: Cycles::new(per_replica_allowance), + refunded_cycles: Cycles::zero(), + refunding_nodes: BTreeSet::new(), + }, + registry_version: RegistryVersion::from(1), + } + } + + fn flexible(n: usize) -> Replication { + let committee: BTreeSet = (0..n as u64) + .map(|i| NodeId::from(PrincipalId::new_node_test_id(i))) + .collect(); + Replication::Flexible { + committee, + min_responses: 1, + max_responses: n as u32, + } + } + + /// Builds a tracker on a `Normal` cost schedule with the given subnet size. + fn make_tracker(ctx: &CanisterHttpRequestContext, subnet_size: u32) -> PayAsYouGoTracker { + PayAsYouGoTracker::new( + ctx, + NumberOfNodes::from(subnet_size), + CanisterCyclesCostSchedule::Normal, + ) + } + + /// Asserts that a gossiping outcall (flexible or non-replicated) charges the + /// gossip term over the full subnet size. + fn assert_gossip_charged_over_subnet_size(replication: Replication) { + let subnet_size = 13_u32; + let ctx = context(replication, 1_000_000_000); + let mut tracker = make_tracker(&ctx, subnet_size); + + let transformed_size = 500_u64; + assert_eq!( + tracker.subtract_gossip_usage(NumBytes::from(transformed_size)), + Ok(()) + ); + let expected = + FLEXIBLE_PER_TRANSFORMED_BYTE_NODE_FEE * transformed_size as u128 * subnet_size as u128; + assert_eq!(tracker.spent, expected); + } + + #[test] + fn does_not_charge_base_cost() { + // The base cost is handled at context creation, so a freshly created + // tracker has spent nothing and a zero-usage request refunds everything. + let ctx = context(Replication::FullyReplicated, 1_000_000); + let tracker = make_tracker(&ctx, 13); + assert_eq!(tracker.spent, 0); + assert_eq!( + tracker.create_payment_receipt().refund, + Cycles::new(1_000_000) + ); + } + + #[test] + fn charges_per_replica_cost_fully_replicated() { + let allowance = 1_000_000_000_u128; + let ctx = context(Replication::FullyReplicated, allowance); + let mut tracker = make_tracker(&ctx, 13); + + let response_size = 1_000_u64; + let response_ms = 2_000_u128; + assert_eq!( + tracker.subtract_network_usage(NetworkUsage { + response_size: NumBytes::from(response_size), + response_time: Duration::from_millis(response_ms as u64), + }), + Ok(()) + ); + let network = + PER_DOWNLOADED_BYTE_FEE * response_size as u128 + PER_RESPONSE_MS_FEE * response_ms; + + let instructions = 13_000_u64; + assert_eq!( + tracker.subtract_transform_usage(NumInstructions::from(instructions)), + Ok(()) + ); + let transform = instructions as u128 / TRANSFORM_INSTRUCTION_DIVISOR; + + // For fully-replicated requests the gossip term is a + // consensus cost and must not be charged here. + assert_eq!(tracker.subtract_gossip_usage(NumBytes::from(5_000)), Ok(())); + + assert_eq!(tracker.spent, network + transform); + assert_eq!( + tracker.create_payment_receipt().refund, + Cycles::new(allowance - network - transform) + ); + } + + #[test] + fn charges_gossip_usage_for_flexible() { + assert_gossip_charged_over_subnet_size(flexible(13)); + } + + #[test] + fn charges_gossip_usage_for_non_replicated() { + // Non-replicated outcalls use the same (flexible) pricing as flexible + // outcalls, so the gossip term is charged over the full subnet size. + let node = NodeId::from(PrincipalId::new_node_test_id(0)); + assert_gossip_charged_over_subnet_size(Replication::NonReplicated(node)); + } + + #[test] + fn returns_pricing_error_when_budget_is_exceeded() { + let ctx = context(Replication::FullyReplicated, 100); + let mut tracker = make_tracker(&ctx, 13); + assert_eq!( + tracker.subtract_network_usage(NetworkUsage { + response_size: NumBytes::from(1_000), + response_time: Duration::ZERO, + }), + Err(PricingError::InsufficientCycles) + ); + assert_eq!(tracker.create_payment_receipt().refund, Cycles::zero()); + } + + #[test] + fn charges_nothing_on_free_cost_schedule() { + // On a free subnet the tracker charges nothing, even for usage that + // would otherwise exceed the allowance, so the full allowance is + // refunded. A flexible request is used so the gossip term (which would + // not be charged for fully-replicated requests) is also exercised. + let allowance = 1_000_000_u128; + let ctx = context(flexible(13), allowance); + let mut tracker = PayAsYouGoTracker::new( + &ctx, + NumberOfNodes::from(13), + CanisterCyclesCostSchedule::Free, + ); + + assert_eq!( + tracker.subtract_network_usage(NetworkUsage { + response_size: NumBytes::from(1_000_000), + response_time: Duration::from_secs(30), + }), + Ok(()) + ); + assert_eq!( + tracker.subtract_transform_usage(NumInstructions::from(1_000_000_000)), + Ok(()) + ); + assert_eq!( + tracker.subtract_gossip_usage(NumBytes::from(1_000_000)), + Ok(()) + ); + + assert_eq!(tracker.spent, 0); + assert_eq!( + tracker.create_payment_receipt().refund, + Cycles::new(allowance) + ); + } +} diff --git a/rs/pocket_ic_server/src/pocket_ic.rs b/rs/pocket_ic_server/src/pocket_ic.rs index 2d76130d4af3..944cea3d1294 100644 --- a/rs/pocket_ic_server/src/pocket_ic.rs +++ b/rs/pocket_ic_server/src/pocket_ic.rs @@ -117,7 +117,8 @@ use ic_types::messages::{ CertificateDelegationFormat, CertificateDelegationMetadata, SignedSenderInfo, }; use ic_types::{ - CanisterId, Height, NumInstructions, PrincipalId, RegistryVersion, SnapshotId, SubnetId, + CanisterId, Height, NumInstructions, NumberOfNodes, PrincipalId, RegistryVersion, SnapshotId, + SubnetId, artifact::UnvalidatedArtifactMutation, canister_http::{ CanisterHttpPaymentReceipt, CanisterHttpReject, @@ -3791,6 +3792,8 @@ impl Operation for ProcessCanisterHttpInternal { id, context, socks_proxy_addrs: vec![], + cost_schedule: CanisterCyclesCostSchedule::Normal, + subnet_size: NumberOfNodes::from(sm.nodes.len() as u32), }) { canister_http.pending.insert(id); } @@ -3965,6 +3968,8 @@ fn process_mock_canister_https_response( id: canister_http_request_id, context: context.clone(), socks_proxy_addrs: vec![], + cost_schedule: CanisterCyclesCostSchedule::Normal, + subnet_size: NumberOfNodes::from(subnet.nodes.len() as u32), }) .unwrap(); let response = loop { diff --git a/rs/types/cycles/src/cycles_cost_schedule.rs b/rs/types/cycles/src/cycles_cost_schedule.rs index b9f4bb119e28..738dbd603710 100644 --- a/rs/types/cycles/src/cycles_cost_schedule.rs +++ b/rs/types/cycles/src/cycles_cost_schedule.rs @@ -3,7 +3,9 @@ use serde::{Deserialize, Serialize}; /// How to charge canisters for their use of computational resources (such as /// executing instructions, storing data, network, etc.) -#[derive(Clone, Copy, Debug, Default, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)] +#[derive( + Clone, Copy, Debug, Default, Serialize, Deserialize, PartialEq, Eq, Hash, PartialOrd, Ord, +)] pub enum CanisterCyclesCostSchedule { #[default] Normal, diff --git a/rs/types/types/src/canister_http.rs b/rs/types/types/src/canister_http.rs index ae794d288131..90dcb2d838c3 100644 --- a/rs/types/types/src/canister_http.rs +++ b/rs/types/types/src/canister_http.rs @@ -42,7 +42,7 @@ //! the timestamp of a request plus the timeout interval. This condition is verifiable by the other nodes in the network. //! Once a timeout has made it into a finalized block, the request is answered with an error message. use crate::{ - CanisterId, CountBytes, RegistryVersion, ReplicaVersion, Time, + CanisterId, CountBytes, NumberOfNodes, RegistryVersion, ReplicaVersion, Time, artifact::{CanisterHttpResponseId, IdentifiableArtifact, PbArtifact}, crypto::{BasicSigOf, CryptoHashOf}, messages::{CallbackId, RejectContext, Request}, @@ -62,7 +62,7 @@ use ic_protobuf::{ proxy::{ProxyDecodeError, try_from_option_field}, state::system_metadata::v1 as pb_metadata, }; -use ic_types_cycles::Cycles; +use ic_types_cycles::{CanisterCyclesCostSchedule, Cycles}; use rand::RngCore; use rand::seq::IteratorRandom; use serde::{Deserialize, Serialize}; @@ -90,6 +90,9 @@ pub const MAX_CANISTER_HTTP_REQUEST_BYTES: u64 = 2_000_000; /// Maximum number of response bytes for a canister http request. pub const MAX_CANISTER_HTTP_RESPONSE_BYTES: u64 = 2_000_000; +/// Maximum size of a canister http reject message. +pub const MAXIMUM_CANISTER_HTTP_ERROR_MESSAGE_BYTES: usize = 1024; // 1KB + /// Maximum number of bytes to represent URL for a canister http request. pub const MAX_CANISTER_HTTP_URL_SIZE: usize = 8192; @@ -181,6 +184,35 @@ pub enum Replication { }, } +impl Replication { + /// Returns the [`ReplicationKind`] of this request. + pub fn kind(&self) -> ReplicationKind { + match self { + Replication::FullyReplicated => ReplicationKind::FullyReplicated, + Replication::Flexible { .. } => ReplicationKind::Flexible, + Replication::NonReplicated(_) => ReplicationKind::NonReplicated, + } + } +} + +/// The kind of replication of a request. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ReplicationKind { + FullyReplicated, + Flexible, + NonReplicated, +} + +impl ReplicationKind { + pub fn as_str(&self) -> &'static str { + match self { + ReplicationKind::FullyReplicated => "fully_replicated", + ReplicationKind::Flexible => "flexible", + ReplicationKind::NonReplicated => "non_replicated", + } + } +} + #[derive(Clone, Eq, PartialEq, Hash, Debug, Deserialize, Serialize, FromRepr)] #[repr(u32)] pub enum PricingVersion { @@ -802,6 +834,12 @@ pub struct CanisterHttpRequest { /// The addresses should be sent in the following format: `socks5://[]:`, for example: /// `socks5://[2602:fb2b:110:10:506f:cff:feff:fe69]:1080` pub socks_proxy_addrs: Vec, + /// The subnet's cycles cost schedule in effect at the request context's + /// registry version. + pub cost_schedule: CanisterCyclesCostSchedule, + /// The number of nodes on the subnet at the request context's registry + /// version. + pub subnet_size: NumberOfNodes, } /// The content of a response after the transformation