feat: CON-1639 HTTPS outcalls pay-as-you-go and dark launch budget trackers (#10519)

## Background
Currently, HTTPS outcalls are charged upfront based on a
`max_response_bytes` parameter. This is done by subtracting the full
cost from the caller's payment. The remaining cycles are stored in the
request context and refunded once a response is delivered.

Instead, we want to introduce pay-as-you-go pricing, which charges
cycles whenever resources are consumed. This happens in three stages:
1. Base cost. This fee is charged for every request upfront
2. Per-replica cost. The remaining cycles after charging the base cost
are split evenly between the participating replicas. Each replica
consumes some of their allowance as the HTTP request is processed. In
the end, the remaining cycles are gossiped as part of refund shares.
3. Consensus cost. This fee is charged for including the aggregated HTTP
response as part of a block. The cost is covered by the sum of refund
shares that were included in the aggregated response. Cycles remaining
after charging the consensus cost are refunded to the user
asynchronously.

The per-replica cost (2.) is calculated by the "budget tracker": This
struct is instantiated with the per-replica cycles allowance, whenever a
new request is starting to be processed by the HTTP adapter. As the
response is downloaded, transformed and gossiped, the tracker charges
the consumed cycles from the initial allowance. If at any point the
remaining allowance _does not_ cover an outstanding charge, an error is
returned and a reject response is gossiped. If the initial allowance
_does_ cover all charges, the remaining cycles are refunded. To this
end, the budget tracker creates a payment receipt which is gossiped
alongside the response.

Before this PR only one budget tracker exited (`LegacyTracker`) which
doesn't compute any per-replica cost, and also doesn't refund anything.

## Proposed Changes
This PR introduces the "pay-as-you-go" budget tracker, whose purpose is
to calculate the per-replica cost for outcalls using the "pay-as-you-go"
pricing (note that such outcalls do not exist yet). This is done by
implementing the per-replica part of the pricing formula defined
[here](https://colab.research.google.com/drive/1MyZBivzU_5kQtqQ8Z71gYbrMubGrTXmv#scrollTo=vxa3d-NYyEOi)
(internal). To do this, we additionally pass the subnet's size and cycle
cost schedule to the HTTP adapter. This is needed to calculate the
correct pricing.

"Pay-as-you-go" pricing charges for the amount of bytes that are
gossiped explicitly in the case of flexible and non-replicated outcalls
(where the whole response is gossiped). To do this, we move the
truncation of oversized reject messages into the adapter, such that the
correct payload length is charged.

Additionally, we implement and start to use a `DarkLaunchTracker`. This
tracker calculates both, the real (legacy, i.e. 0) and the new
(pay-as-you-go) per-replica cost. In the end, only the "real" refund is
gossiped. However, this allows us to compare both trackers, and observe
whenever the pay-as-you-go tracker returns an out-of-cycles error, while
the legacy tracker succeeds. Such an event indicates that the outcall
would not be covered by enough cycles under the new pricing. In this
case, the canister ID is logged and a metric increased.

The legacy charging flow should be unchanged by this PR.

---------

Co-authored-by: IDX GitHub Automation <infra+github-automation@dfinity.org>

Author: eichhorl

Cargo.lock

@@ -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",

rs/https_outcalls/client/BUILD.bazel

@@ -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 }

rs/https_outcalls/client/Cargo.toml

@@ -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,13 +82,15 @@ 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,
             tx,
             rx,
             query_service,
             metrics,
+            pricing_factory,
             log,
         }
     }
@@ -121,6 +126,7 @@ impl NonBlockingChannel<CanisterHttpRequest> 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<CanisterHttpRequest> 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<CanisterHttpRequest> 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<CanisterHttpRequest> 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<CanisterHttpRequest> 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<CanisterHttpRequest> 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<CanisterHttpRequest> 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() {

rs/https_outcalls/client/src/client.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],
             ),
         }
     }

rs/https_outcalls/client/src/metrics.rs

@@ -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",

rs/https_outcalls/consensus/BUILD.bazel

@@ -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 }