diff --git a/rs/consensus/src/consensus/metrics.rs b/rs/consensus/src/consensus/metrics.rs index 4e79485e9e02..da3a46065bf5 100644 --- a/rs/consensus/src/consensus/metrics.rs +++ b/rs/consensus/src/consensus/metrics.rs @@ -188,6 +188,7 @@ pub(crate) struct FinalizerMetrics { pub canister_http_timeouts_delivered: IntCounter, pub canister_http_divergences_delivered: IntCounter, pub canister_http_out_of_cycles_delivered: IntCounter, + pub canister_http_async_receipts_delivered: IntCounter, pub canister_http_flexible_candid_failures: IntCounter, pub canister_http_flexible_errors_delivered: IntCounter, pub canister_http_payload_bytes_delivered: Histogram, @@ -304,6 +305,10 @@ impl FinalizerMetrics { "canister_http_out_of_cycles_delivered", "Total number of canister http messages delivered as out of cycles", ), + canister_http_async_receipts_delivered: metrics_registry.int_counter( + "canister_http_async_receipts_delivered", + "Total number of canister http asynchronous receipts delivered", + ), canister_http_flexible_candid_failures: metrics_registry.int_counter( "canister_http_flexible_candid_failures", "Total number of flexible canister http responses skipped due to candid encoding/decoding failures", @@ -371,6 +376,8 @@ impl FinalizerMetrics { .inc_by(batch_stats.canister_http.divergence_responses as u64); self.canister_http_out_of_cycles_delivered .inc_by(batch_stats.canister_http.out_of_cycles as u64); + self.canister_http_async_receipts_delivered + .inc_by(batch_stats.canister_http.async_receipts as u64); let flexible_ok_candid_failures = batch_stats .canister_http diff --git a/rs/https_outcalls/consensus/benches/payload_validation.rs b/rs/https_outcalls/consensus/benches/payload_validation.rs index 235251f056bf..da4c73b222a1 100644 --- a/rs/https_outcalls/consensus/benches/payload_validation.rs +++ b/rs/https_outcalls/consensus/benches/payload_validation.rs @@ -1,14 +1,14 @@ //! Benchmark for the validation of canister HTTP outcall //! payloads. -use std::collections::{BTreeMap, BTreeSet, HashSet}; +use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; use criterion::{BenchmarkId, Criterion, black_box, criterion_group, criterion_main}; use ic_consensus_mocks::{Dependencies, DependenciesBuilder}; use ic_crypto_temp_crypto::{NodeKeysToGenerate, TempCryptoComponent}; -use ic_https_outcalls_consensus::payload_builder::CanisterHttpPayloadBuilderImpl; +use ic_https_outcalls_consensus::payload_builder::{CanisterHttpPayloadBuilderImpl, PastPayloads}; use ic_https_outcalls_pricing::fees::{flexible_initial_spent, non_flexible_initial_spent}; use ic_interfaces::crypto::BasicSigner; use ic_interfaces_registry::RegistryClient; @@ -148,7 +148,7 @@ fn bench_payload_verification(c: &mut Criterion) { black_box(target.builder.validate_canister_http_payload_impl( black_box(&target.payload), black_box(&target.validation_context), - black_box(HashSet::new()), + black_box(PastPayloads::default()), )) .expect("validation failed"); }) @@ -470,6 +470,7 @@ impl<'a> PayloadAssembler<'a> { divergence_responses, flexible_responses, flexible_errors: vec![], + async_receipts: vec![], }; assert!( diff --git a/rs/https_outcalls/consensus/src/gossip.rs b/rs/https_outcalls/consensus/src/gossip.rs index b7eee0c6f012..070b71fff835 100644 --- a/rs/https_outcalls/consensus/src/gossip.rs +++ b/rs/https_outcalls/consensus/src/gossip.rs @@ -37,8 +37,13 @@ impl BouncerFactory &latest_state.get_ref().metadata.subnet_call_context_manager; let known_request_ids: BTreeSet<_> = subnet_call_context_manger .canister_http_request_contexts - .iter() - .map(|item| *item.0) + .keys() + .chain( + subnet_call_context_manger + .delivered_canister_http_request_contexts + .keys(), + ) + .copied() .collect(); let next_callback_id = subnet_call_context_manger.next_callback_id(); (known_request_ids, next_callback_id) @@ -72,3 +77,145 @@ impl BouncerFactory std::time::Duration::from_secs(3) } } + +#[cfg(test)] +mod tests { + use super::*; + use ic_artifact_pool::canister_http_pool::CanisterHttpPoolImpl; + use ic_interfaces_state_manager::Labeled; + use ic_logger::replica_logger::no_op_logger; + use ic_metrics::MetricsRegistry; + use ic_registry_subnet_type::SubnetType; + use ic_replicated_state::metadata_state::subnet_call_context_manager::SubnetCallContext; + use ic_test_utilities::state_manager::RefMockStateManager; + use ic_test_utilities_types::{ + ids::{node_test_id, subnet_test_id}, + messages::RequestBuilder, + }; + use ic_types::{ + Height, NumberOfNodes, RegistryVersion, ReplicaVersion, + canister_http::{ + CanisterHttpMethod, CanisterHttpPaymentReceipt, CanisterHttpRequestContext, + CanisterHttpResponseMetadata, CanisterHttpResponseReceipt, PricingVersion, + RefundStatus, Replication, + }, + crypto::{BasicSig, BasicSigOf, CryptoHash, CryptoHashOf, Signed}, + signature::BasicSignature, + time::UNIX_EPOCH, + }; + use ic_types_cycles::CanisterCyclesCostSchedule; + use std::sync::Arc; + + fn request_context() -> CanisterHttpRequestContext { + CanisterHttpRequestContext { + request: RequestBuilder::new().build(), + url: String::new(), + max_response_bytes: None, + headers: vec![], + body: None, + http_method: CanisterHttpMethod::GET, + transform: None, + time: UNIX_EPOCH, + replication: Replication::FullyReplicated, + pricing_version: PricingVersion::PayAsYouGo, + refund_status: RefundStatus::default(), + registry_version: RegistryVersion::from(1), + subnet_size: NumberOfNodes::from(13), + cost_schedule: CanisterCyclesCostSchedule::Normal, + } + } + + fn share_id(callback_id: CallbackId) -> CanisterHttpResponseId { + Signed { + content: CanisterHttpResponseReceipt { + metadata: CanisterHttpResponseMetadata { + id: callback_id, + content_hash: CryptoHashOf::new(CryptoHash(vec![])), + content_size: 0, + is_reject: false, + replica_version: ReplicaVersion::default(), + }, + payment_receipt: CanisterHttpPaymentReceipt::default(), + }, + signature: BasicSignature { + signature: BasicSigOf::new(BasicSig(vec![])), + signer: node_test_id(0), + }, + } + } + + /// The `next_callback_id` of the state [`test_bouncer`] builds. + const NEXT_CALLBACK_ID: u64 = 3; + + /// A bouncer over a state that has handed out callback ids 0, 1 and 2 — so + /// `next_callback_id` is [`NEXT_CALLBACK_ID`] — of which 0 is still awaiting a + /// response, 1 has been responded to, and 2 is gone for good. + fn test_bouncer() -> Bouncer { + let mut state = ReplicatedState::new(subnet_test_id(0), SubnetType::Application); + let contexts = &mut state.metadata.subnet_call_context_manager; + // Advance `next_callback_id` to NEXT_CALLBACK_ID + for _ in 0..NEXT_CALLBACK_ID { + contexts.push_context(SubnetCallContext::CanisterHttpRequest(request_context())); + } + contexts.canister_http_request_contexts.clear(); + contexts + .canister_http_request_contexts + .insert(CallbackId::new(0), request_context()); + contexts + .delivered_canister_http_request_contexts + .insert(CallbackId::new(1), request_context()); + + let state_manager = Arc::new(RefMockStateManager::default()); + state_manager + .get_mut() + .expect_get_latest_state() + .return_const(Labeled::new(Height::new(1), Arc::new(state))); + + let gossip = CanisterHttpGossipImpl::new(state_manager); + let pool = CanisterHttpPoolImpl::new(MetricsRegistry::new(), no_op_logger()); + gossip.new_bouncer(&pool) + } + + #[test] + fn shares_of_delivered_contexts_are_wanted() { + let bouncer = test_bouncer(); + + assert_eq!(bouncer(&share_id(CallbackId::new(0))), BouncerValue::Wants); + assert_eq!(bouncer(&share_id(CallbackId::new(1))), BouncerValue::Wants); + // A request that is neither pending nor delivered is settled for good. + assert_eq!( + bouncer(&share_id(CallbackId::new(2))), + BouncerValue::Unwanted + ); + } + + #[test] + fn shares_of_upcoming_requests_are_wanted() { + let bouncer = test_bouncer(); + + // The very next id execution will hand out, ... + assert_eq!( + bouncer(&share_id(CallbackId::new(NEXT_CALLBACK_ID))), + BouncerValue::Wants + ); + // ... and everything up to the far edge of the look-ahead window. + assert_eq!( + bouncer(&share_id(CallbackId::new( + NEXT_CALLBACK_ID + MAX_NUMBER_OF_REQUESTS_AHEAD + ))), + BouncerValue::Wants + ); + } + + #[test] + fn shares_beyond_the_look_ahead_window_are_stashed() { + let bouncer = test_bouncer(); + + assert_eq!( + bouncer(&share_id(CallbackId::new( + NEXT_CALLBACK_ID + MAX_NUMBER_OF_REQUESTS_AHEAD + 1 + ))), + BouncerValue::MaybeWantsLater + ); + } +} diff --git a/rs/https_outcalls/consensus/src/payload_builder.rs b/rs/https_outcalls/consensus/src/payload_builder.rs index fcb8e9ba5f32..98fb55a55a11 100644 --- a/rs/https_outcalls/consensus/src/payload_builder.rs +++ b/rs/https_outcalls/consensus/src/payload_builder.rs @@ -5,8 +5,8 @@ use crate::{ payload_builder::{ parse::bytes_to_payload, utils::{ - FlexibleFindResult, ResponseShareSigInput, find_flexible_result, - find_fully_replicated_response, find_non_flexible_out_of_cycles, + FlexibleFindResult, RefundedNodes, ResponseShareSigInput, find_async_receipts, + find_flexible_result, find_fully_replicated_response, find_non_flexible_out_of_cycles, find_non_replicated_response, group_shares_by_callback_id, grouped_shares_meet_divergence_criteria, response_share_sig_inputs, validate_flexible_response_with_proof, validate_response_share, @@ -16,7 +16,7 @@ use crate::{ use candid::{Decode, Encode}; use ic_consensus_utils::{ crypto::ConsensusCrypto, - membership::{CanisterHttpCommittee, Membership}, + membership::{CanisterHttpCommittee, Membership, MembershipError}, }; use ic_error_types::RejectCode; use ic_https_outcalls_pricing::fees::{flexible_initial_spent, non_flexible_initial_spent}; @@ -41,18 +41,20 @@ use ic_management_canister_types_private::{ use ic_metrics::MetricsRegistry; use ic_registry_client_helpers::subnet::SubnetRegistry; use ic_replicated_state::ReplicatedState; +use ic_replicated_state::metadata_state::subnet_call_context_manager::DELIVERED_CANISTER_HTTP_REQUEST_CONTEXT_TIMEOUT; use ic_types::{ CountBytes, Height, NodeId, NumBytes, RegistryVersion, SubnetId, batch::{ - CanisterHttpInitialSpent, CanisterHttpPayload, CanisterHttpSpent, ConsensusResponse, - FlexibleCanisterHttpError, FlexibleCanisterHttpResponseWithProof, + CanisterHttpAsyncSpent, CanisterHttpInitialSpent, CanisterHttpPayload, CanisterHttpSpent, + ConsensusResponse, FlexibleCanisterHttpError, FlexibleCanisterHttpResponseWithProof, FlexibleCanisterHttpResponses, MAX_CANISTER_HTTP_PAYLOAD_SIZE, ValidationContext, }, canister_http::{ CANISTER_HTTP_MAX_RESPONSES_PER_BLOCK, CANISTER_HTTP_TIMEOUT_INTERVAL, - CanisterHttpResponseContent, CanisterHttpResponseDivergence, CanisterHttpResponseShare, - Replication, + CanisterHttpRequestContext, CanisterHttpResponseContent, CanisterHttpResponseDivergence, + CanisterHttpResponseShare, Replication, }, + consensus::Threshold, messages::{CallbackId, Payload, RejectContext}, registry::RegistryClientError, signature::BasicSigBatchEntry, @@ -64,6 +66,8 @@ use std::{ }; pub(crate) mod parse; +pub use parse::PastPayloads; + #[cfg(all(test, feature = "proptest"))] mod proptests; #[cfg(test)] @@ -79,6 +83,7 @@ pub struct CanisterHttpBatchStats { pub timeouts: usize, pub divergence_responses: usize, pub out_of_cycles: usize, + pub async_receipts: usize, pub single_signature_responses: usize, pub flexible_ok_responses: usize, pub flexible_ok_responses_candid_failures: usize, @@ -138,7 +143,7 @@ impl CanisterHttpPayloadBuilderImpl { fn get_canister_http_payload_impl( &self, validation_context: &ValidationContext, - delivered_ids: HashSet, + past_payloads: PastPayloads, max_payload_size: NumBytes, ) -> CanisterHttpPayload { let state = match self @@ -156,11 +161,16 @@ impl CanisterHttpPayloadBuilderImpl { } }; - let canister_http_request_contexts = &state - .get_ref() - .metadata - .subnet_call_context_manager - .canister_http_request_contexts; + let PastPayloads { + delivered_ids, + refunded_nodes, + } = past_payloads; + + let subnet_call_context_manager = &state.get_ref().metadata.subnet_call_context_manager; + let canister_http_request_contexts = + &subnet_call_context_manager.canister_http_request_contexts; + let delivered_canister_http_request_contexts = + &subnet_call_context_manager.delivered_canister_http_request_contexts; let mut accumulated_size = 0; let mut responses_included = 0; @@ -171,6 +181,7 @@ impl CanisterHttpPayloadBuilderImpl { let mut out_of_cycles = vec![]; let mut flexible_responses = vec![]; let mut flexible_errors = vec![]; + let mut async_receipts = vec![]; // Metrics counters let mut total_share_count = 0; @@ -374,6 +385,44 @@ impl CanisterHttpPayloadBuilderImpl { }, } } + + // Collect the asynchronous receipts of the requests that have already + // been responded to. + for (callback_id, request) in delivered_canister_http_request_contexts { + if responses_included >= CANISTER_HTTP_MAX_RESPONSES_PER_BLOCK { + // Break early to avoid iterating through all open contexts. + break; + } + // Skip contexts that have already timed out. + if delivered_context_timed_out(request, validation_context) { + continue; + } + let Some(grouped_shares) = shares_by_callback_id.get(callback_id) else { + continue; + }; + let committee = match self.request_committee(request) { + Ok(committee) => committee, + Err(err) => { + warn!(self.log, "Failed to get canister http committee: {:?}", err); + continue; + } + }; + // Skip shares for nodes that have already issued a refund for this request, + // according to the certified state or any past payload above it. + let already_refunded = RefundedNodes::new(*callback_id, request, &refunded_nodes); + for share in find_async_receipts(grouped_shares, &committee, &already_refunded) { + if responses_included >= CANISTER_HTTP_MAX_RESPONSES_PER_BLOCK { + break; + } + let share_size = share.count_bytes(); + let size = NumBytes::new((accumulated_size + share_size) as u64); + if size < max_payload_size { + async_receipts.push(share.clone()); + responses_included += 1; + accumulated_size += share_size; + } + } + } } CanisterHttpPayload { @@ -383,6 +432,57 @@ impl CanisterHttpPayloadBuilderImpl { out_of_cycles, flexible_responses, flexible_errors, + async_receipts, + } + } + + /// The set of replicas that may have produced a receipt for this request, + /// evaluated at the registry version pinned in the request context. + fn request_committee( + &self, + context: &CanisterHttpRequestContext, + ) -> Result, MembershipError> { + match &context.replication { + Replication::FullyReplicated => self + .membership + .get_canister_http_committee(context.registry_version) + .map(|committee| BTreeSet::from_iter(committee.committee)), + // Only the designated replica ever produces a receipt. + Replication::NonReplicated(node_id) => Ok(BTreeSet::from([*node_id])), + Replication::Flexible { committee, .. } => Ok(committee.clone()), + } + } + + /// The replicas a response to a *non-flexible* request may come from, and how + /// many of them have to agree on one for it to be delivered. + fn non_flexible_committee( + &self, + callback_id: CallbackId, + context: &CanisterHttpRequestContext, + ) -> Result<(BTreeSet, Threshold), CanisterHttpPayloadValidationError> { + match &context.replication { + Replication::FullyReplicated => { + let CanisterHttpCommittee { + committee, + threshold, + .. + } = self + .membership + .get_canister_http_committee(context.registry_version) + .map_err(|err| { + warn!(self.log, "Failed to get membership: {:?}", err); + CanisterHttpPayloadValidationError::ValidationFailed( + CanisterHttpPayloadValidationFailure::Membership, + ) + })?; + Ok((BTreeSet::from_iter(committee), threshold)) + } + Replication::NonReplicated(node_id) => Ok((BTreeSet::from([*node_id]), 1)), + Replication::Flexible { .. } => { + Err(CanisterHttpPayloadValidationError::InvalidArtifact( + InvalidCanisterHttpPayloadReason::InvalidPayloadSection(callback_id), + )) + } } } @@ -390,8 +490,12 @@ impl CanisterHttpPayloadBuilderImpl { &self, payload: &CanisterHttpPayload, validation_context: &ValidationContext, - mut delivered_ids: HashSet, + past_payloads: PastPayloads, ) -> Result<(), PayloadValidationError> { + let PastPayloads { + mut delivered_ids, + refunded_nodes, + } = past_payloads; // Empty payloads are always valid if payload.is_empty() { return Ok(()); @@ -423,11 +527,10 @@ impl CanisterHttpPayloadBuilderImpl { CanisterHttpPayloadValidationFailure::StateUnavailable, ) })?; - let http_contexts = &state - .get_ref() - .metadata - .subnet_call_context_manager - .canister_http_request_contexts; + let subnet_call_context_manager = &state.get_ref().metadata.subnet_call_context_manager; + let http_contexts = &subnet_call_context_manager.canister_http_request_contexts; + let delivered_http_contexts = + &subnet_call_context_manager.delivered_canister_http_request_contexts; // Validate the timed out calls for timeout_id in &payload.timeouts { @@ -487,50 +590,26 @@ impl CanisterHttpPayloadBuilderImpl { .map_err(CanisterHttpPayloadValidationError::InvalidArtifact)?; let subnet_size = request_context.subnet_size; - let (effective_committee, effective_threshold) = match request_context.replication { - Replication::NonReplicated(node_id) => (vec![node_id], 1), - Replication::FullyReplicated => { - // The committee is the subnet node set at the registry - // version pinned in the request context. - let CanisterHttpCommittee { - committee, - threshold, - .. - } = self - .membership - .get_canister_http_committee(request_context.registry_version) - .map_err(|err| { - warn!(self.log, "Failed to get membership: {:?}", err); - CanisterHttpPayloadValidationError::ValidationFailed( - CanisterHttpPayloadValidationFailure::Membership, - ) - })?; - (committee, threshold) - } - Replication::Flexible { .. } => { - return invalid_artifact( - InvalidCanisterHttpPayloadReason::InvalidPayloadSection(callback_id), - ); - } - }; + let (effective_committee, effective_threshold) = + self.non_flexible_committee(callback_id, request_context)?; let (valid_signers, invalid_signers): (Vec, Vec) = response .proof .signatures .keys() .cloned() - .partition(|signer| effective_committee.iter().any(|id| id == signer)); + .partition(|signer| effective_committee.contains(signer)); if !invalid_signers.is_empty() { return invalid_artifact(InvalidCanisterHttpPayloadReason::SignersNotMembers { invalid_signers, - committee: effective_committee, + committee: effective_committee.into_iter().collect(), valid_signers, }); } if valid_signers.len() < effective_threshold { return invalid_artifact(InvalidCanisterHttpPayloadReason::NotEnoughSigners { - committee: effective_committee, + committee: effective_committee.into_iter().collect(), signers: valid_signers, expected_threshold: effective_threshold, }); @@ -691,32 +770,7 @@ impl CanisterHttpPayloadBuilderImpl { )?; // Which replicas a response could come from, and how many of them have to // agree on it for it to be delivered. - let (committee, threshold) = match &context.replication { - Replication::FullyReplicated => { - let CanisterHttpCommittee { - committee, - threshold, - .. - } = self - .membership - .get_canister_http_committee(context.registry_version) - .map_err(|err| { - warn!(self.log, "Failed to get membership: {:?}", err); - CanisterHttpPayloadValidationError::ValidationFailed( - CanisterHttpPayloadValidationFailure::Membership, - ) - })?; - (BTreeSet::from_iter(committee), threshold) - } - // Only the designated replica's response is ever delivered, so it is a - // committee of one and its own threshold. - Replication::NonReplicated(node_id) => (BTreeSet::from([*node_id]), 1), - Replication::Flexible { .. } => { - return invalid_artifact( - InvalidCanisterHttpPayloadReason::InvalidPayloadSection(callback_id), - ); - } - }; + let (committee, threshold) = self.non_flexible_committee(callback_id, context)?; let mut seen_signers = HashSet::new(); for share in &error.shares { @@ -1103,6 +1157,56 @@ impl CanisterHttpPayloadBuilderImpl { } } + // Validate asynchronous receipts: the signed spends of replicas that the + // already delivered response of their outcall did not account for. + let mut receipts_by_callback: BTreeMap> = + BTreeMap::new(); + for share in &payload.async_receipts { + receipts_by_callback + .entry(share.content.id()) + .or_default() + .push(share); + } + for (callback_id, shares) in receipts_by_callback { + // Only an outcall that has already been responded to can be refunded asynchronously. + let context = delivered_http_contexts.get(&callback_id).ok_or( + CanisterHttpPayloadValidationError::InvalidArtifact( + InvalidCanisterHttpPayloadReason::UnknownDeliveredCallbackId(callback_id), + ), + )?; + // Reject if the context for this share has already timed out. + if delivered_context_timed_out(context, validation_context) { + return invalid_artifact( + InvalidCanisterHttpPayloadReason::DeliveredCallbackTimedOut(callback_id), + ); + } + let committee = self.request_committee(context).map_err(|err| { + warn!(self.log, "Failed to get membership: {:?}", err); + CanisterHttpPayloadValidationError::ValidationFailed( + CanisterHttpPayloadValidationFailure::Membership, + ) + })?; + + // A replica may only be refunded once. + let already_refunded = RefundedNodes::new(callback_id, context, &refunded_nodes); + let mut seen_signers = HashSet::new(); + for &share in &shares { + validate_response_share(share, callback_id, &committee, &mut seen_signers, context) + .map_err(CanisterHttpPayloadValidationError::InvalidArtifact)?; + + let signer = share.signature.signer; + if already_refunded.contains(&signer) { + return invalid_artifact(InvalidCanisterHttpPayloadReason::AlreadyRefunded { + callback_id, + signer, + }); + } + } + + // Defer signature verification. + sig_inputs.extend(response_share_sig_inputs(shares, context.registry_version)); + } + // Batch-verify the signatures of the deferred shares. if !sig_inputs.is_empty() { self.crypto @@ -1147,8 +1251,8 @@ impl BatchPayloadBuilder for CanisterHttpPayloadBuilderImpl { max_size, NumBytes::new(MAX_CANISTER_HTTP_PAYLOAD_SIZE as u64), ); - let delivered_ids = parse::parse_past_payload_ids(past_payloads, &self.log); - let payload = self.get_canister_http_payload_impl(context, delivered_ids, max_size); + let past_payloads = parse::parse_past_payloads(past_payloads, &self.log); + let payload = self.get_canister_http_payload_impl(context, past_payloads, max_size); parse::payload_to_bytes(payload, max_size) } @@ -1181,7 +1285,7 @@ impl BatchPayloadBuilder for CanisterHttpPayloadBuilderImpl { )); } - let delivered_ids = parse::parse_past_payload_ids(past_payloads, &self.log); + let past_payloads = parse::parse_past_payloads(past_payloads, &self.log); let payload = parse::bytes_to_payload(payload).map_err(|e| { ValidationError::InvalidArtifact( consensus::InvalidPayloadReason::InvalidCanisterHttpPayload( @@ -1192,7 +1296,7 @@ impl BatchPayloadBuilder for CanisterHttpPayloadBuilderImpl { self.validate_canister_http_payload_impl( &payload, proposal_context.validation_context, - delivered_ids, + past_payloads, ) } } @@ -1404,6 +1508,20 @@ impl } } + let mut async_spent: BTreeMap> = BTreeMap::new(); + for share in messages.async_receipts { + stats.async_receipts += 1; + async_spent + .entry(share.content.id()) + .or_default() + .insert(share.signature.signer, share.content.spent()); + } + spent.asynchronous.extend( + async_spent + .into_iter() + .map(|(callback, shares)| CanisterHttpAsyncSpent { callback, shares }), + ); + (consensus_responses, spent, stats) } } @@ -1663,6 +1781,18 @@ fn divergence_response_into_reject( )) } +/// Returns true if a delivered context has timed out, meaning no further +/// asynchronous receipts are accepted. +fn delivered_context_timed_out( + context: &CanisterHttpRequestContext, + validation_context: &ValidationContext, +) -> bool { + validation_context + .time + .saturating_duration_since(context.time) + >= DELIVERED_CANISTER_HTTP_REQUEST_CONTEXT_TIMEOUT +} + fn validation_failed( err: CanisterHttpPayloadValidationFailure, ) -> Result<(), PayloadValidationError> { diff --git a/rs/https_outcalls/consensus/src/payload_builder/parse.rs b/rs/https_outcalls/consensus/src/payload_builder/parse.rs index 13c1e0692584..b3b9d65bb1e7 100644 --- a/rs/https_outcalls/consensus/src/payload_builder/parse.rs +++ b/rs/https_outcalls/consensus/src/payload_builder/parse.rs @@ -6,14 +6,15 @@ use ic_protobuf::{ types::v1::{CanisterHttpResponseMessage, canister_http_response_message::MessageType}, }; use ic_types::{ - NumBytes, + NodeId, NumBytes, PrincipalId, batch::{ CanisterHttpOutOfCycles, CanisterHttpPayload, FlexibleCanisterHttpError, FlexibleCanisterHttpResponses, iterator_to_bytes, slice_to_messages, }, + canister_http::CanisterHttpResponseShare, messages::CallbackId, }; -use std::collections::HashSet; +use std::collections::{BTreeMap, HashSet}; pub(crate) fn bytes_to_payload(data: &[u8]) -> Result { let messages: Vec = @@ -36,6 +37,9 @@ pub(crate) fn bytes_to_payload(data: &[u8]) -> Result payload .out_of_cycles .push(CanisterHttpOutOfCycles::try_from(out_of_cycles)?), + Some(MessageType::AsyncReceipt(share)) => payload + .async_receipts + .push(CanisterHttpResponseShare::try_from(share)?), None => return Err(ProxyDecodeError::MissingField("message_type")), } } @@ -51,6 +55,7 @@ pub(crate) fn payload_to_bytes(payload: CanisterHttpPayload, max_size: NumBytes) responses, flexible_responses, flexible_errors, + async_receipts, } = payload; let message_iterator = @@ -101,33 +106,75 @@ pub(crate) fn payload_to_bytes(payload: CanisterHttpPayload, max_size: NumBytes) pb::CanisterHttpOutOfCycles::from(out_of_cycles), )), }), + ) + .chain( + async_receipts + .into_iter() + .map(|share| CanisterHttpResponseMessage { + message_type: Some(MessageType::AsyncReceipt(pb::CanisterHttpShare::from( + share, + ))), + }), ); iterator_to_bytes(message_iterator, max_size) } -pub(crate) fn parse_past_payload_ids( +/// Relevant data of payloads between the certified height and the block being built. +#[derive(Default)] +pub struct PastPayloads { + /// The callback ids that have already been responded to. + pub delivered_ids: HashSet, + /// Per callback id, the replicas whose spend has already been reported + /// asynchronously, i.e. that must not be refunded again. + pub refunded_nodes: BTreeMap>, +} + +/// Collects from the `past_payloads` everything a new payload must not repeat: +/// the responses already delivered and the asynchronous receipts already +/// reported. +pub(crate) fn parse_past_payloads( past_payloads: &[PastPayload], log: &ReplicaLogger, -) -> HashSet { - past_payloads - .iter() - .flat_map(|payload| { - slice_to_messages::(payload.payload).unwrap_or_else( - |err| { - error!( - log, - "Failed to parse CanisterHttp past payload for height {}. Error: {}", - payload.height, - err - ); - vec![] - }, - ) - }) - .filter_map(get_id_from_message) - .map(CallbackId::new) - .collect() +) -> PastPayloads { + let mut parsed = PastPayloads::default(); + for payload in past_payloads { + let messages = slice_to_messages::(payload.payload) + .unwrap_or_else(|err| { + error!( + log, + "Failed to parse CanisterHttp past payload for height {}. Error: {}", + payload.height, + err + ); + vec![] + }); + for message in messages { + if let Some(MessageType::AsyncReceipt(share)) = &message.message_type { + if let Some((callback_id, signer)) = callback_and_signer_of_share(share) { + parsed + .refunded_nodes + .entry(callback_id) + .or_default() + .insert(signer); + } + continue; + } + if let Some(id) = get_id_from_message(message) { + parsed.delivered_ids.insert(CallbackId::new(id)); + } + } + } + parsed +} + +/// Extracts the callback and signer IDs of a [`pb::CanisterHttpShare`], or +/// `None` if either is missing or malformed. Such a share would have failed +/// payload validation, so it cannot appear in a past payload. +fn callback_and_signer_of_share(share: &pb::CanisterHttpShare) -> Option<(CallbackId, NodeId)> { + let callback_id = CallbackId::new(share.metadata.as_ref()?.id); + let signer = PrincipalId::try_from(share.signature.as_ref()?.signer.as_slice()).ok()?; + Some((callback_id, NodeId::from(signer))) } /// Extracts the CallbackId (as u64) from a [`CanisterHttpResponseMessage`] @@ -144,6 +191,8 @@ fn get_id_from_message(message: CanisterHttpResponseMessage) -> Option { Some(MessageType::FlexibleError(flex_error)) => Some(flex_error.callback_id), Some(MessageType::OutOfCycles(out_of_cycles)) => Some(out_of_cycles.callback_id), Some(MessageType::Timeout(id)) => Some(id), + // Handled by `parse_past_payloads`, which does not deliver a response for it. + Some(MessageType::AsyncReceipt(_)) => None, None => None, } } diff --git a/rs/https_outcalls/consensus/src/payload_builder/tests.rs b/rs/https_outcalls/consensus/src/payload_builder/tests.rs index d84edac3ea4f..70b5ecb19fcb 100644 --- a/rs/https_outcalls/consensus/src/payload_builder/tests.rs +++ b/rs/https_outcalls/consensus/src/payload_builder/tests.rs @@ -35,6 +35,7 @@ use ic_management_canister_types_private::{ }; use ic_metrics::MetricsRegistry; use ic_registry_subnet_features::SubnetFeatures; +use ic_replicated_state::metadata_state::subnet_call_context_manager::DELIVERED_CANISTER_HTTP_REQUEST_CONTEXT_TIMEOUT; use ic_test_utilities::state_manager::RefMockStateManager; use ic_test_utilities_consensus::fake::FakeContentSigner; use ic_test_utilities_registry::SubnetRecordBuilder; @@ -213,6 +214,7 @@ fn multiple_payload_test() { divergence_responses: vec![], flexible_responses: vec![], flexible_errors: vec![], + async_receipts: vec![], }; let past_payload = payload_to_bytes(past_payload, TEST_MAX_PAYLOAD_BYTES); @@ -510,6 +512,7 @@ fn divergence_responses_count_toward_max_responses() { divergence_responses, flexible_responses: vec![], flexible_errors: vec![], + async_receipts: vec![], }; let result = payload_builder.validate_payload( @@ -663,6 +666,7 @@ fn duplicate_validation() { divergence_responses: vec![], flexible_responses: vec![], flexible_errors: vec![], + async_receipts: vec![], }; let payload = payload_to_bytes(payload, TEST_MAX_PAYLOAD_BYTES); let past_payloads = vec![PastPayload { @@ -720,6 +724,7 @@ fn divergence_response_validation_test() { }], flexible_responses: vec![], flexible_errors: vec![], + async_receipts: vec![], }; let payload = payload_to_bytes(payload, TEST_MAX_PAYLOAD_BYTES); @@ -743,6 +748,7 @@ fn divergence_response_validation_test() { }], flexible_responses: vec![], flexible_errors: vec![], + async_receipts: vec![], }; let payload = payload_to_bytes(payload, TEST_MAX_PAYLOAD_BYTES); @@ -783,6 +789,7 @@ fn divergence_response_validation_test() { }], flexible_responses: vec![], flexible_errors: vec![], + async_receipts: vec![], }; let payload = payload_to_bytes(payload, TEST_MAX_PAYLOAD_BYTES); @@ -841,6 +848,7 @@ fn divergence_duplicate_signer_rejected() { }], flexible_responses: vec![], flexible_errors: vec![], + async_receipts: vec![], }; let payload = payload_to_bytes(payload, TEST_MAX_PAYLOAD_BYTES); @@ -2068,6 +2076,7 @@ where divergence_responses: vec![], flexible_responses: vec![], flexible_errors: vec![], + async_receipts: vec![], }; let payload = payload_to_bytes(payload, TEST_MAX_PAYLOAD_BYTES); @@ -3377,6 +3386,7 @@ fn into_messages_emits_initial_spend_reports() { timeouts: vec![timeout_callback], divergence_responses: vec![divergence], out_of_cycles: vec![out_of_cycles], + async_receipts: vec![], }; let bytes = payload_to_bytes_max_4mb(payload); @@ -7511,6 +7521,28 @@ fn setup_test_with_contexts( }); } +/// Like [`setup_test_with_contexts`], but the contexts are injected as already +/// responded to, i.e. as delivered contexts awaiting their asynchronous receipts. +fn setup_test_with_delivered_contexts( + num_nodes: usize, + delivered: Vec<(CallbackId, CanisterHttpRequestContext)>, + run: impl FnOnce(CanisterHttpPayloadBuilderImpl, Arc>), +) { + // The context's subnet size must match the subnet the test registers, since + // payload building and validation source the subnet size from the context. + let delivered: Vec<_> = delivered + .into_iter() + .map(|(cb, mut ctx)| { + ctx.subnet_size = NumberOfNodes::from(num_nodes as u32); + (cb, ctx) + }) + .collect(); + test_config_with_http_feature(true, num_nodes, |mut payload_builder, pool| { + inject_contexts(&mut payload_builder, [], delivered, None); + run(payload_builder, pool); + }); +} + fn setup_test_with_flexible_context( num_nodes: usize, callback_id: CallbackId, @@ -7545,17 +7577,40 @@ pub(crate) fn inject_request_contexts_with_cost_schedule( payload_builder: &mut CanisterHttpPayloadBuilderImpl, contexts: impl IntoIterator, cost_schedule: Option, +) { + inject_contexts(payload_builder, contexts, [], cost_schedule); +} + +/// Replaces the payload_builder's state_reader with one containing the given +/// request contexts, split into the ones still awaiting a response and the +/// `delivered` ones, which have already been responded to and are only kept +/// around for their asynchronous receipts. +pub(crate) fn inject_contexts( + payload_builder: &mut CanisterHttpPayloadBuilderImpl, + contexts: impl IntoIterator, + delivered: impl IntoIterator, + cost_schedule: Option, ) { let mut init_state = ic_test_utilities_state::get_initial_state(0, 0); - for (cb, mut ctx) in contexts { + let with_cost_schedule = |mut ctx: CanisterHttpRequestContext| { if let Some(cost_schedule) = cost_schedule { ctx.cost_schedule = cost_schedule; } + ctx + }; + for (cb, ctx) in contexts { init_state .metadata .subnet_call_context_manager .canister_http_request_contexts - .insert(cb, ctx); + .insert(cb, with_cost_schedule(ctx)); + } + for (cb, ctx) in delivered { + init_state + .metadata + .subnet_call_context_manager + .delivered_canister_http_request_contexts + .insert(cb, with_cost_schedule(ctx)); } let state_manager = Arc::new(RefMockStateManager::default()); state_manager @@ -7773,6 +7828,7 @@ fn flexible_payload(groups: Vec) -> CanisterHttpP divergence_responses: vec![], flexible_responses: groups, flexible_errors: vec![], + async_receipts: vec![], } } @@ -7893,3 +7949,630 @@ fn mock_crypto_rejecting_signatures() -> MockCrypto { }); mock_crypto } + +// =================================================================== +// Asynchronous receipts +// =================================================================== + +/// A delivered (already responded to) pay-as-you-go context whose replicas each +/// got `TEST_PER_REPLICA_ALLOWANCE`, with `refunding_nodes` already accounted for +/// by the response that was delivered. +fn delivered_context( + replication: Replication, + refunding_nodes: impl IntoIterator, +) -> CanisterHttpRequestContext { + let mut context = with_payg_allowance(request_context(replication), TEST_PER_REPLICA_ALLOWANCE); + context.refund_status.refunding_nodes = refunding_nodes.into_iter().collect(); + context +} + +/// The signers of the payload's asynchronous receipts, all of which must be for +/// `callback_id`. +fn async_receipt_signers( + payload: &CanisterHttpPayload, + callback_id: CallbackId, +) -> BTreeSet { + payload + .async_receipts + .iter() + .map(|share| { + assert_eq!(share.content.id(), callback_id); + share.signature.signer + }) + .collect() +} + +/// Builds a payload against a delivered context of the given `replication` whose +/// response was signed by `refunding_nodes`, with `shares_in_pool` many replicas +/// having produced a share, and returns it. +fn build_async_receipt_payload( + num_nodes: usize, + callback_id: CallbackId, + replication: Replication, + refunding_nodes: impl IntoIterator, + shares_in_pool: usize, + past_payloads: &[PastPayload], +) -> CanisterHttpPayload { + let (response, metadata) = test_response_and_metadata(callback_id.get()); + let shares = metadata_to_shares(shares_in_pool, &metadata); + let mut payload = None; + setup_test_with_delivered_contexts( + num_nodes, + vec![(callback_id, delivered_context(replication, refunding_nodes))], + |payload_builder, canister_http_pool| { + { + let mut pool_access = canister_http_pool.write().unwrap(); + if let Some(own) = shares.first() { + add_own_share_to_pool(pool_access.deref_mut(), own, &response); + add_received_shares_to_pool(pool_access.deref_mut(), shares[1..].to_vec()); + } + } + let context = default_validation_context(); + let bytes = payload_builder.build_payload( + Height::new(1), + TEST_MAX_PAYLOAD_BYTES, + past_payloads, + &context, + ); + assert_matches!( + payload_builder.validate_payload( + Height::new(1), + &test_proposal_context(&context), + &bytes, + past_payloads, + ), + Ok(()) + ); + payload = Some(bytes_to_payload(&bytes).expect("parse error")); + }, + ); + payload.expect("payload was not built") +} + +/// A delivered context is only reported on until it times out: from then on message +/// routing settles whatever is left of the allowances — refunding every replica that +/// has not been accounted for in full — and drops the context, so a receipt would +/// only be spending block space on cycles that are refunded anyway. +/// +/// The boundary matters: message routing applies the receipts of a block before +/// timing out the contexts, so a receipt included exactly at the timeout would +/// still be applied, which is precisely what the payload builder avoids here. +#[test] +fn async_receipt_is_not_reported_once_the_delivered_context_times_out() { + let num_nodes = 4; + let callback_id = CallbackId::new(0); + let (response, metadata) = test_response_and_metadata(callback_id.get()); + let shares = metadata_to_shares(num_nodes, &metadata); + + // The context is stamped at `UNIX_EPOCH`, so the block time alone decides + // whether it has timed out. Check either side of the boundary, so that the + // negative case cannot pass just because nothing was refundable anyway. + for (elapsed, expect_refunds) in [ + ( + DELIVERED_CANISTER_HTTP_REQUEST_CONTEXT_TIMEOUT - Duration::from_nanos(1), + true, + ), + (DELIVERED_CANISTER_HTTP_REQUEST_CONTEXT_TIMEOUT, false), + ] { + setup_test_with_delivered_contexts( + num_nodes, + vec![( + callback_id, + delivered_context(Replication::FullyReplicated, []), + )], + |payload_builder, canister_http_pool| { + { + let mut pool_access = canister_http_pool.write().unwrap(); + add_own_share_to_pool(pool_access.deref_mut(), &shares[0], &response); + add_received_shares_to_pool(pool_access.deref_mut(), shares[1..].to_vec()); + } + let context = ValidationContext { + time: UNIX_EPOCH + elapsed, + ..default_validation_context() + }; + let payload = + build_and_validate_and_parse_payload_with_context(&payload_builder, &context); + assert_eq!( + !payload.async_receipts.is_empty(), + expect_refunds, + "unexpected refunds {:?} after {elapsed:?}", + payload.async_receipts, + ); + }, + ); + } +} + +/// A replica is left out for either of two reasons: it was already accounted for by +/// the delivered response, or it never produced a receipt at all — the latter is +/// refunded in full when the delivered context times out instead. +#[test] +fn async_receipt_reports_the_replicas_that_answered_and_are_unaccounted_for() { + let num_nodes = 4; + let callback_id = CallbackId::new(0); + // Node 0 signed the response that was delivered; nodes 1 and 2 answered too + // late to be part of it; node 3 never answered at all. + let payload = build_async_receipt_payload( + num_nodes, + callback_id, + Replication::FullyReplicated, + [node_test_id(0)], + /* shares_in_pool = */ 3, + &[], + ); + + assert_eq!( + async_receipt_signers(&payload, callback_id), + BTreeSet::from([node_test_id(1), node_test_id(2)]) + ); +} + +/// Once every replica has been accounted for, there is nothing left to report. +#[test] +fn async_receipt_is_not_reported_when_every_replica_has_been_accounted_for() { + let num_nodes = 4; + let callback_id = CallbackId::new(0); + let all_nodes: Vec<_> = (0..num_nodes as u64).map(node_test_id).collect(); + + let payload = build_async_receipt_payload( + num_nodes, + callback_id, + Replication::FullyReplicated, + all_nodes, + num_nodes, + &[], + ); + + assert!(payload.async_receipts.is_empty(), "{payload:?}"); + // A payload with nothing to report is empty, i.e. not put into the block at all. + assert!(payload.is_empty(), "{payload:?}"); +} + +/// A refund reported by a payload above the certified height is not repeated, even +/// though the certified state does not reflect it yet. +#[test] +fn async_receipt_is_not_repeated_after_a_past_payload_reported_it() { + let num_nodes = 4; + let callback_id = CallbackId::new(0); + let (_, metadata) = test_response_and_metadata(callback_id.get()); + + // A past payload already reported node 2. + let past = payload_to_bytes_max_4mb(CanisterHttpPayload { + async_receipts: vec![metadata_to_share(2, &metadata)], + ..Default::default() + }); + let past_payloads = vec![PastPayload { + height: Height::new(1), + time: UNIX_EPOCH, + block_hash: CryptoHashOf::from(CryptoHash(vec![])), + payload: &past, + }]; + + let payload = build_async_receipt_payload( + num_nodes, + callback_id, + Replication::FullyReplicated, + [node_test_id(0)], + num_nodes, + &past_payloads, + ); + + // Node 0 was accounted for by the delivered response, node 2 by the past + // payload; only nodes 1 and 3 are left. + assert_eq!( + async_receipt_signers(&payload, callback_id), + BTreeSet::from([node_test_id(1), node_test_id(3)]) + ); +} + +/// The replicas an asynchronous receipt may come from are the ones the request +/// pins: the whole subnet for a fully replicated request, the designated replica +/// for a non-replicated one, and the recorded committee for a flexible one. +#[test] +fn async_receipts_are_reported_for_the_committee_of_every_replication() { + let num_nodes = 4; + let callback_id = CallbackId::new(0); + + for (replication, refunding_nodes, expected) in [ + // The response was signed by node 0; every other replica of the subnet that + // answered is still unaccounted for. + ( + Replication::FullyReplicated, + vec![node_test_id(0)], + BTreeSet::from([node_test_id(1), node_test_id(2), node_test_id(3)]), + ), + // The request timed out before the designated replica answered, so nobody is + // accounted for yet — and only that replica can report anything. + ( + Replication::NonReplicated(node_test_id(2)), + vec![], + BTreeSet::from([node_test_id(2)]), + ), + // Node 3 is not part of the flexible committee, so its share is ignored even + // though it is in the pool. + ( + Replication::Flexible { + committee: BTreeSet::from([node_test_id(0), node_test_id(1), node_test_id(2)]), + min_responses: 2, + max_responses: 3, + }, + vec![node_test_id(0)], + BTreeSet::from([node_test_id(1), node_test_id(2)]), + ), + ] { + let payload = build_async_receipt_payload( + num_nodes, + callback_id, + replication.clone(), + refunding_nodes, + /* shares_in_pool = */ num_nodes, + &[], + ); + + assert_eq!( + async_receipt_signers(&payload, callback_id), + expected, + "unexpected receipts for {replication:?}" + ); + } +} + +/// Receipts count towards the per-block response limit like any other message, and +/// are collected across delivered contexts until it is reached. +#[test] +fn async_receipts_of_many_delivered_contexts_stop_at_the_per_block_limit() { + // Enough delivered contexts, each answered by every replica, to produce more + // receipts than a single block may carry. The subnet size deliberately does not + // divide the limit, so that it is reached part-way through a context. + let num_nodes = 3; + let num_contexts = CANISTER_HTTP_MAX_RESPONSES_PER_BLOCK / num_nodes + 1; + let contexts: Vec<_> = (0..num_contexts as u64) + .map(|id| { + ( + CallbackId::new(id), + delivered_context(Replication::FullyReplicated, []), + ) + }) + .collect(); + + setup_test_with_delivered_contexts(num_nodes, contexts, |payload_builder, pool| { + { + let mut pool_access = pool.write().unwrap(); + for id in 0..num_contexts as u64 { + let (_, metadata) = test_response_and_metadata(id); + add_received_shares_to_pool( + pool_access.deref_mut(), + metadata_to_shares(num_nodes, &metadata), + ); + } + } + + let payload = build_and_validate_and_parse_payload(&payload_builder); + + assert_eq!( + payload.async_receipts.len(), + CANISTER_HTTP_MAX_RESPONSES_PER_BLOCK + ); + // The last context that fits is only partly reported, i.e. collecting stops + // mid-context rather than at a context boundary. + assert_eq!( + payload + .async_receipts + .iter() + .map(|share| share.content.id()) + .collect::>() + .len(), + CANISTER_HTTP_MAX_RESPONSES_PER_BLOCK.div_ceil(num_nodes) + ); + }); +} + +/// An asynchronous receipt reported for a callback that has not been responded to is +/// invalid: only a delivered context can be refunded this way. +#[test] +fn validate_payload_fails_for_an_async_receipt_of_an_unanswered_request() { + let num_nodes = 4; + let callback_id = CallbackId::new(0); + let (_, metadata) = test_response_and_metadata(callback_id.get()); + let payload = CanisterHttpPayload { + async_receipts: vec![metadata_to_share(0, &metadata)], + ..Default::default() + }; + + let mut result = None; + setup_test_with_contexts( + num_nodes, + vec![( + callback_id, + delivered_context(Replication::FullyReplicated, []), + )], + |payload_builder, _pool| { + result = Some(payload_builder.validate_payload( + Height::new(1), + &test_proposal_context(&default_validation_context()), + &payload_to_bytes_max_4mb(payload), + &[], + )); + }, + ); + + assert_matches!( + result.expect("validation did not run"), + Err(ValidationError::InvalidArtifact( + InvalidPayloadReason::InvalidCanisterHttpPayload( + InvalidCanisterHttpPayloadReason::UnknownDeliveredCallbackId(id), + ), + )) if id == callback_id + ); +} + +/// Validates a payload carrying nothing but the given asynchronous receipts, +/// against a delivered request with the given `context` and the given `past_payloads`. +fn validate_async_receipt_payload( + num_nodes: usize, + callback_id: CallbackId, + context: CanisterHttpRequestContext, + refunds: Vec, + past_payloads: &[PastPayload], +) -> Result<(), PayloadValidationError> { + validate_async_receipt_payload_at( + num_nodes, + callback_id, + context, + refunds, + past_payloads, + &default_validation_context(), + ) +} + +/// Same as [`validate_async_receipt_payload`], but for a block proposed in the +/// given `validation_context` rather than the default one. +fn validate_async_receipt_payload_at( + num_nodes: usize, + callback_id: CallbackId, + context: CanisterHttpRequestContext, + refunds: Vec, + past_payloads: &[PastPayload], + validation_context: &ValidationContext, +) -> Result<(), PayloadValidationError> { + let payload = CanisterHttpPayload { + async_receipts: refunds, + ..Default::default() + }; + let mut result = None; + setup_test_with_delivered_contexts( + num_nodes, + vec![(callback_id, context)], + |payload_builder, _pool| { + result = Some(payload_builder.validate_payload( + Height::new(1), + &test_proposal_context(validation_context), + &payload_to_bytes_max_4mb(payload), + past_payloads, + )); + }, + ); + result.expect("validation did not run") +} + +#[test] +fn validate_payload_fails_for_an_async_receipt_of_a_timed_out_delivered_context() { + let callback_id = CallbackId::new(0); + let (_, metadata) = test_response_and_metadata(callback_id.get()); + + // The context is stamped at `UNIX_EPOCH`, so the block time alone decides + // whether it has timed out. + let validate_at = |elapsed| { + validate_async_receipt_payload_at( + 4, + callback_id, + delivered_context(Replication::FullyReplicated, []), + vec![metadata_to_share(1, &metadata)], + &[], + &ValidationContext { + time: UNIX_EPOCH + elapsed, + ..default_validation_context() + }, + ) + }; + + // Just short of the timeout the very same receipt is still accepted, so the + // rejection below cannot be down to anything else about it. + assert_matches!( + validate_at(DELIVERED_CANISTER_HTTP_REQUEST_CONTEXT_TIMEOUT - Duration::from_nanos(1)), + Ok(()) + ); + assert_matches!( + validate_at(DELIVERED_CANISTER_HTTP_REQUEST_CONTEXT_TIMEOUT), + Err(ValidationError::InvalidArtifact( + InvalidPayloadReason::InvalidCanisterHttpPayload( + InvalidCanisterHttpPayloadReason::DeliveredCallbackTimedOut(id), + ), + )) if id == callback_id + ); +} + +fn assert_already_refunded(result: Result<(), PayloadValidationError>, expected_signer: NodeId) { + assert_matches!( + result, + Err(ValidationError::InvalidArtifact( + InvalidPayloadReason::InvalidCanisterHttpPayload( + InvalidCanisterHttpPayloadReason::AlreadyRefunded { signer, .. }, + ), + )) if signer == expected_signer + ); +} + +/// A replica already accounted for by the delivered response must not be refunded +/// a second time. +#[test] +fn validate_payload_fails_for_an_async_receipt_of_an_accounted_replica() { + let callback_id = CallbackId::new(0); + let (_, metadata) = test_response_and_metadata(callback_id.get()); + + let result = validate_async_receipt_payload( + 4, + callback_id, + delivered_context(Replication::FullyReplicated, [node_test_id(1)]), + vec![metadata_to_share(1, &metadata)], + &[], + ); + + assert_already_refunded(result, node_test_id(1)); +} + +/// A replica already reported by a payload above the certified height must not be +/// refunded again, even though the certified state does not reflect it yet. +#[test] +fn validate_payload_fails_for_an_async_receipt_repeated_from_a_past_payload() { + let callback_id = CallbackId::new(0); + let (_, metadata) = test_response_and_metadata(callback_id.get()); + let refund = metadata_to_share(1, &metadata); + let past = payload_to_bytes_max_4mb(CanisterHttpPayload { + async_receipts: vec![refund.clone()], + ..Default::default() + }); + let past_payloads = vec![PastPayload { + height: Height::new(1), + time: UNIX_EPOCH, + block_hash: CryptoHashOf::from(CryptoHash(vec![])), + payload: &past, + }]; + + let result = validate_async_receipt_payload( + 4, + callback_id, + delivered_context(Replication::FullyReplicated, []), + vec![refund], + &past_payloads, + ); + + assert_already_refunded(result, node_test_id(1)); +} + +/// The same replica must not be refunded twice by the same payload, whether the two +/// receipts are byte-identical or merely share a signer. +#[test] +fn validate_payload_fails_for_an_async_receipt_repeated_within_the_payload() { + let callback_id = CallbackId::new(0); + let (_, metadata) = test_response_and_metadata(callback_id.get()); + let share = metadata_to_share(1, &metadata); + let same_signer = metadata_to_share_with_spent(1, &metadata, Cycles::new(1)); + + for refunds in [vec![share.clone(), share.clone()], vec![share, same_signer]] { + let result = validate_async_receipt_payload( + 4, + callback_id, + delivered_context(Replication::FullyReplicated, []), + refunds, + &[], + ); + + assert_matches!( + result, + Err(ValidationError::InvalidArtifact( + InvalidPayloadReason::InvalidCanisterHttpPayload( + InvalidCanisterHttpPayloadReason::DuplicateShareSigner { signer, .. }, + ), + )) if signer == node_test_id(1) + ); + } +} + +/// Only the request's committee can be refunded for it. +#[test] +fn validate_payload_fails_for_an_async_receipt_of_a_non_committee_replica() { + let callback_id = CallbackId::new(0); + let (_, metadata) = test_response_and_metadata(callback_id.get()); + let designated = node_test_id(0); + let outsider = node_test_id(1); + + let result = validate_async_receipt_payload( + 4, + callback_id, + // Only the designated replica of a non-replicated request ever spends. + delivered_context(Replication::NonReplicated(designated), []), + vec![metadata_to_share(node_id_to_u64(outsider), &metadata)], + &[], + ); + + assert_matches!( + result, + Err(ValidationError::InvalidArtifact( + InvalidPayloadReason::InvalidCanisterHttpPayload( + InvalidCanisterHttpPayloadReason::ShareSignerNotInCommittee { signer, .. }, + ), + )) if signer == outsider + ); +} + +/// A receipt claiming a spend beyond the replica's allowance is rejected here as +/// everywhere else. +#[test] +fn validate_payload_fails_for_an_async_receipt_with_an_excessive_spend() { + let callback_id = CallbackId::new(0); + let (_, metadata) = test_response_and_metadata(callback_id.get()); + + let result = validate_async_receipt_payload( + 4, + callback_id, + delivered_context(Replication::FullyReplicated, []), + vec![metadata_to_share_with_spent( + 1, + &metadata, + TEST_PER_REPLICA_ALLOWANCE + Cycles::new(1), + )], + &[], + ); + + assert_matches!( + result, + Err(ValidationError::InvalidArtifact( + InvalidPayloadReason::InvalidCanisterHttpPayload( + InvalidCanisterHttpPayloadReason::SpentExceedsLimit { .. }, + ), + )) + ); +} + +/// An asynchronous receipt reports what its replicas spent, without delivering a +/// response: the response it belongs to is already out. Receipts are regrouped by +/// the callback they are signed for, one report per callback. +#[test] +fn into_messages_emits_asynchronous_spend_reports() { + let (first, second) = (CallbackId::new(7), CallbackId::new(8)); + let (_, first_metadata) = test_response_and_metadata(first.get()); + let (_, second_metadata) = test_response_and_metadata(second.get()); + let spent = Cycles::new(1_234); + let payload = CanisterHttpPayload { + // Deliberately interleaved, to show that the grouping does not rely on order. + async_receipts: vec![ + metadata_to_share_with_spent(1, &first_metadata, spent), + metadata_to_share_with_spent(1, &second_metadata, Cycles::new(7)), + metadata_to_share_with_spent(2, &first_metadata, Cycles::zero()), + ], + ..Default::default() + }; + + let (responses, spent_report, stats) = + CanisterHttpPayloadBuilderImpl::into_messages(&payload_to_bytes_max_4mb(payload)); + + assert!(responses.is_empty(), "{responses:?}"); + assert_eq!(stats.async_receipts, 3); + assert!(spent_report.initial.is_empty()); + let reported: BTreeMap<_, _> = spent_report + .asynchronous + .iter() + .map(|report| (report.callback, report.shares.clone())) + .collect(); + assert_eq!( + reported, + BTreeMap::from([ + ( + first, + BTreeMap::from([(node_test_id(1), spent), (node_test_id(2), Cycles::zero())]) + ), + (second, BTreeMap::from([(node_test_id(1), Cycles::new(7))])), + ]) + ); +} diff --git a/rs/https_outcalls/consensus/src/payload_builder/utils.rs b/rs/https_outcalls/consensus/src/payload_builder/utils.rs index 0c96fe061b7f..f98a2fccc041 100644 --- a/rs/https_outcalls/consensus/src/payload_builder/utils.rs +++ b/rs/https_outcalls/consensus/src/payload_builder/utils.rs @@ -339,6 +339,46 @@ pub(crate) fn find_non_flexible_out_of_cycles( }) } +/// The replicas of an already responded to outcall whose spend has been accounted +/// for, and that must therefore not be refunded a second time. +pub(crate) struct RefundedNodes<'a> { + by_context: &'a BTreeSet, + by_past_payload: Option<&'a HashSet>, +} + +impl<'a> RefundedNodes<'a> { + pub(crate) fn new( + callback_id: CallbackId, + context: &'a CanisterHttpRequestContext, + past_payload_refunds: &'a BTreeMap>, + ) -> Self { + Self { + by_context: &context.refund_status.refunding_nodes, + by_past_payload: past_payload_refunds.get(&callback_id), + } + } + + pub(crate) fn contains(&self, node_id: &NodeId) -> bool { + self.by_context.contains(node_id) + || self + .by_past_payload + .is_some_and(|nodes| nodes.contains(node_id)) + } +} + +/// The receipts of the replicas that have contributed to an already responded to +/// outcall but have not been refunded yet, at most one per replica. +pub(crate) fn find_async_receipts<'a>( + grouped_shares: &BTreeMap>, + committee: &BTreeSet, + already_refunded: &RefundedNodes, +) -> Vec<&'a CanisterHttpResponseShare> { + one_share_per_committee_member(grouped_shares, committee) + .into_iter() + .filter(|share| !already_refunded.contains(&share.signature.signer)) + .collect() +} + /// Reconstructs, for every signer of an aggregated proof, the /// [`CanisterHttpResponseShare`] that signer actually signed: the shared /// [`CanisterHttpResponseMetadata`] combined with that signer's own diff --git a/rs/https_outcalls/consensus/src/pool_manager.rs b/rs/https_outcalls/consensus/src/pool_manager.rs index 9952ed700ae0..84d9c5546801 100644 --- a/rs/https_outcalls/consensus/src/pool_manager.rs +++ b/rs/https_outcalls/consensus/src/pool_manager.rs @@ -112,9 +112,11 @@ impl CanisterHttpPoolManagerImpl { } } - /// Purge shares of responses for requests that have already been processed. + /// Purge shares of responses for requests that have already been processed, + /// i.e. whose contexts are no longer part of the replicated state. fn purge_shares_of_processed_requests( &self, + state: &ReplicatedState, canister_http_pool: &dyn CanisterHttpPool, ) -> CanisterHttpChangeSet { let _time = self @@ -123,13 +125,16 @@ impl CanisterHttpPoolManagerImpl { .with_label_values(&["purge_shares"]) .start_timer(); - let active_callback_ids = self.active_callback_ids(); - let next_callback_id = self.next_callback_id(); + let known_callback_ids = Self::known_callback_ids(state); + let next_callback_id = state + .metadata + .subnet_call_context_manager + .next_callback_id(); let ids_to_remove_from_cache: Vec<_> = self .requested_id_cache .borrow() - .difference(&active_callback_ids) + .difference(&known_callback_ids) .cloned() .collect(); @@ -140,7 +145,7 @@ impl CanisterHttpPoolManagerImpl { canister_http_pool .get_validated_shares() .filter_map(|share| { - if active_callback_ids.contains(&share.content.id()) { + if known_callback_ids.contains(&share.content.id()) { None } else { Some(CanisterHttpChangeAction::RemoveValidated(share.clone())) @@ -153,7 +158,7 @@ impl CanisterHttpPoolManagerImpl { .filter(|artifact| artifact.share.content.id() < next_callback_id) .filter_map(|artifact| { let share = &artifact.share; - if active_callback_ids.contains(&share.content.id()) { + if known_callback_ids.contains(&share.content.id()) { None } else { Some(CanisterHttpChangeAction::RemoveUnvalidated(share.clone())) @@ -164,7 +169,7 @@ impl CanisterHttpPoolManagerImpl { canister_http_pool .get_response_content_items() .filter_map(|content| { - if active_callback_ids.contains(&content.1.id) { + if known_callback_ids.contains(&content.1.id) { None } else { Some(CanisterHttpChangeAction::RemoveContent(content.0.clone())) @@ -255,15 +260,18 @@ impl CanisterHttpPoolManagerImpl { } /// Inform the HttpAdapterShim of any new requests that must be made. - fn make_new_requests(&self, canister_http_pool: &dyn CanisterHttpPool) { + fn make_new_requests( + &self, + state: &ReplicatedState, + canister_http_pool: &dyn CanisterHttpPool, + ) { let _time = self .metrics .op_duration .with_label_values(&["make_new_requests"]) .start_timer(); - let http_requests = &self - .latest_state() + let http_requests = &state .metadata .subnet_call_context_manager .canister_http_request_contexts; @@ -323,7 +331,7 @@ impl CanisterHttpPoolManagerImpl { /// Create any shares that should be made from responses provided by the /// HttpAdapterShim. - fn create_shares_from_responses(&self) -> CanisterHttpChangeSet { + fn create_shares_from_responses(&self, state: &ReplicatedState) -> CanisterHttpChangeSet { let _time = self .metrics .op_duration @@ -331,19 +339,22 @@ impl CanisterHttpPoolManagerImpl { .start_timer(); let mut change_set = Vec::new(); - let active_contexts = &self - .latest_state() - .metadata - .subnet_call_context_manager - .canister_http_request_contexts; + let subnet_call_context_manager = &state.metadata.subnet_call_context_manager; + let active_contexts = &subnet_call_context_manager.canister_http_request_contexts; + let delivered_contexts = + &subnet_call_context_manager.delivered_canister_http_request_contexts; loop { match self.http_adapter_shim.lock().unwrap().try_receive() { Err(TryReceiveError::Empty) => break, 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 { + // Drop the response if its context is no longer present in the replicated state. + // We continue gossiping a share even if a response to the context has already + // been delivered, in order to report the amount of cycles spent. + let Some(context) = active_contexts + .get(&response.id) + .or_else(|| delivered_contexts.get(&response.id)) + else { warn!( self.log, "Dropping http response for request ID {}: \ @@ -410,19 +421,22 @@ impl CanisterHttpPoolManagerImpl { } /// Validate any shares found in the unvalidated section of the canister http pool. - fn validate_shares(&self, canister_http_pool: &dyn CanisterHttpPool) -> CanisterHttpChangeSet { + fn validate_shares( + &self, + state: &ReplicatedState, + canister_http_pool: &dyn CanisterHttpPool, + ) -> CanisterHttpChangeSet { let _time = self .metrics .op_duration .with_label_values(&["validate_shares"]) .start_timer(); - let state = self.latest_state(); - let active_contexts = &state - .metadata - .subnet_call_context_manager - .canister_http_request_contexts; - let next_callback_id = self.next_callback_id(); + let subnet_call_context_manager = &state.metadata.subnet_call_context_manager; + let active_contexts = &subnet_call_context_manager.canister_http_request_contexts; + let delivered_contexts = + &subnet_call_context_manager.delivered_canister_http_request_contexts; + let next_callback_id = subnet_call_context_manager.next_callback_id(); let key_from_share = |share: &CanisterHttpResponseShare| (share.signature.signer, share.content.id()); @@ -450,7 +464,10 @@ impl CanisterHttpPoolManagerImpl { )); } - let Some(context) = active_contexts.get(&share.content.id()) else { + let Some(context) = active_contexts + .get(&share.content.id()) + .or_else(|| delivered_contexts.get(&share.content.id())) + else { return Some(CanisterHttpChangeAction::RemoveUnvalidated(share.clone())); }; @@ -563,19 +580,20 @@ impl CanisterHttpPoolManagerImpl { .with_label_values(&["generate_change_set"]) .start_timer(); let mut change_set = Vec::new(); + let state = self.latest_state(); // Whenever we have artifacts to purge, we insert the purge change actions before everything // else, to avoid having in the validated pool artifacts belonging to different epochs and // hence preserving the expected maximal number of artifacts in the pool. - change_set.extend(self.purge_shares_of_processed_requests(canister_http_pool)); + change_set.extend(self.purge_shares_of_processed_requests(&state, canister_http_pool)); // Make any requests that need to be made and create shares from responses // that are now available. - self.make_new_requests(canister_http_pool); - change_set.extend(self.create_shares_from_responses()); + self.make_new_requests(&state, canister_http_pool); + change_set.extend(self.create_shares_from_responses(&state)); // Attempt to validate unvalidated shares - change_set.extend(self.validate_shares(canister_http_pool)); + change_set.extend(self.validate_shares(&state, canister_http_pool)); self.metrics .in_client_requests @@ -584,14 +602,20 @@ impl CanisterHttpPoolManagerImpl { change_set } - fn active_callback_ids(&self) -> BTreeSet { - self.state_reader - .get_latest_state() - .get_ref() - .metadata - .subnet_call_context_manager + /// The callback ids of all requests whose artifacts are still of use: those + /// still awaiting a response, plus those already responded to but still awaiting + /// the [asynchronous receipts](ic_types::batch::CanisterHttpPayload::async_receipts) + /// of the replicas that did not contribute to the response. + fn known_callback_ids(state: &ReplicatedState) -> BTreeSet { + let subnet_call_context_manager = &state.metadata.subnet_call_context_manager; + subnet_call_context_manager .canister_http_request_contexts .keys() + .chain( + subnet_call_context_manager + .delivered_canister_http_request_contexts + .keys(), + ) .copied() .collect() } @@ -599,15 +623,6 @@ impl CanisterHttpPoolManagerImpl { fn latest_state(&self) -> Arc { self.state_reader.get_latest_state().get_ref().clone() } - - fn next_callback_id(&self) -> CallbackId { - self.state_reader - .get_latest_state() - .get_ref() - .metadata - .subnet_call_context_manager - .next_callback_id() - } } impl PoolMutationsProducer for CanisterHttpPoolManagerImpl { @@ -690,6 +705,27 @@ pub mod test { replicated_state } + /// A state whose HTTP outcall contexts have all been responded to already, i.e. + /// that only keeps them around for their asynchronous receipts. + fn state_with_delivered_http_calls( + delivered: BTreeMap, + ) -> ReplicatedState { + let mut replicated_state = ReplicatedState::new(subnet_test_id(0), SubnetType::System); + let contexts = &mut replicated_state.metadata.subnet_call_context_manager; + // Hand out callback ids up to the largest delivered one, so that shares for + // them are not mistaken for shares belonging to a future state. Pushing a + // context is the only way to advance the private `next_callback_id`, so the + // contexts it parks in the active collection are cleared out again below. + if let Some((max_id, context)) = delivered.iter().next_back() { + for _ in 0..=max_id.get() { + contexts.push_context(SubnetCallContext::CanisterHttpRequest(context.clone())); + } + } + contexts.canister_http_request_contexts.clear(); + contexts.delivered_canister_http_request_contexts = delivered; + replicated_state + } + fn empty_canister_http_response(id: u64) -> CanisterHttpResponse { CanisterHttpResponse { id: CallbackId::from(id), @@ -812,7 +848,8 @@ pub mod test { log, ); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = + pool_manager.validate_shares(&pool_manager.latest_state(), &canister_http_pool); // Make sure the changes are empty (share was filtered out) assert!(changes.is_empty()); @@ -919,7 +956,8 @@ pub mod test { log, ); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = + pool_manager.validate_shares(&pool_manager.latest_state(), &canister_http_pool); assert_matches!(&changes[0], CanisterHttpChangeAction::RemoveUnvalidated(_)); }) @@ -1021,7 +1059,8 @@ pub mod test { log, ); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = + pool_manager.validate_shares(&pool_manager.latest_state(), &canister_http_pool); // The share is dropped silently (removed, not marked invalid). assert_eq!(changes.len(), 1); @@ -1141,7 +1180,8 @@ pub mod test { log, ); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = + pool_manager.validate_shares(&pool_manager.latest_state(), &canister_http_pool); // Make sure the second share is sorted out as invalid, for the right reason. if let CanisterHttpChangeAction::HandleInvalid(_, err) = &changes[0] { @@ -1244,7 +1284,8 @@ pub mod test { timestamp: UNIX_EPOCH, }); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = pool_manager + .validate_shares(&pool_manager.latest_state(), &canister_http_pool); assert_matches!(&changes[0], CanisterHttpChangeAction::HandleInvalid(_, reason) if reason == "Artifact should contain response"); } @@ -1269,7 +1310,8 @@ pub mod test { timestamp: UNIX_EPOCH, }); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = pool_manager + .validate_shares(&pool_manager.latest_state(), &canister_http_pool); assert_matches!( &changes[0], @@ -1298,7 +1340,8 @@ pub mod test { timestamp: UNIX_EPOCH, }); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = pool_manager + .validate_shares(&pool_manager.latest_state(), &canister_http_pool); assert_matches!( &changes[0], @@ -1326,7 +1369,8 @@ pub mod test { timestamp: UNIX_EPOCH, }); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = pool_manager + .validate_shares(&pool_manager.latest_state(), &canister_http_pool); assert_matches!( &changes[0], @@ -1421,7 +1465,8 @@ pub mod test { ); // 4. ACTION: Our replica attempts to validate the artifact. - let change_set = pool_manager.validate_shares(&canister_http_pool); + let change_set = + pool_manager.validate_shares(&pool_manager.latest_state(), &canister_http_pool); // 5. ASSERTION: The artifact must be invalidated with the specific reason. assert_eq!(change_set.len(), 1, "Expected exactly one change action"); @@ -1525,7 +1570,8 @@ pub mod test { log, ); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = + pool_manager.validate_shares(&pool_manager.latest_state(), &canister_http_pool); // The change action should be HandleInvalid because a fully replicated request's // artifact must not contain a response in the unvalidated pool. @@ -1636,7 +1682,8 @@ pub mod test { timestamp: UNIX_EPOCH, }); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = pool_manager + .validate_shares(&pool_manager.latest_state(), &canister_http_pool); // The error from `validate_response_size` itself. let validation_err = format!( @@ -1700,7 +1747,8 @@ pub mod test { timestamp: UNIX_EPOCH, }); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = pool_manager + .validate_shares(&pool_manager.latest_state(), &canister_http_pool); assert_matches!(&changes[0], CanisterHttpChangeAction::MoveToValidated(_)); } @@ -1805,7 +1853,8 @@ pub mod test { }); // 4. VALIDATE: Call validate_shares and check the result. - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = + pool_manager.validate_shares(&pool_manager.latest_state(), &canister_http_pool); // 5. ASSERT: The artifact should be successfully validated and moved to the validated pool. assert_eq!(changes.len(), 1); @@ -1906,7 +1955,8 @@ pub mod test { ); // 4. ACTION: Our replica attempts to validate the artifact. - let change_set = pool_manager.validate_shares(&canister_http_pool); + let change_set = + pool_manager.validate_shares(&pool_manager.latest_state(), &canister_http_pool); // 5. ASSERTION: The artifact is now correctly invalidated by the validate_content_size check. assert_eq!(change_set.len(), 1, "Expected exactly one change action"); @@ -2004,7 +2054,8 @@ pub mod test { log, ); - let change_set = pool_manager.validate_shares(&canister_http_pool); + let change_set = + pool_manager.validate_shares(&pool_manager.latest_state(), &canister_http_pool); let expected_error = format!( "Http Response for request ID {} is too large: Response size {} exceeds the maximum allowed size of {}", @@ -2097,7 +2148,9 @@ pub mod test { .insert(callback_id); assert!( - pool_manager.create_shares_from_responses().is_empty(), + pool_manager + .create_shares_from_responses(&pool_manager.latest_state()) + .is_empty(), "an oversized response must not be signed", ); // The request is deliberately *not* re-requested: the adapter would @@ -2218,7 +2271,8 @@ pub mod test { }); // 3. Call validate_shares and assert that the share is considered VALID. - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = + pool_manager.validate_shares(&pool_manager.latest_state(), &canister_http_pool); assert_matches!(&changes[0], CanisterHttpChangeAction::MoveToValidated(_)); }) @@ -2480,7 +2534,8 @@ pub mod test { .borrow_mut() .insert(stale_callback_id); - let change_set = pool_manager.create_shares_from_responses(); + let change_set = + pool_manager.create_shares_from_responses(&pool_manager.latest_state()); // Only the response for the active context produces a share; the // stale one is dropped. @@ -2575,7 +2630,8 @@ pub mod test { ); // 3. Call the function and get the change set. - let change_set = pool_manager.create_shares_from_responses(); + let change_set = + pool_manager.create_shares_from_responses(&pool_manager.latest_state()); // 4. Assert that the correct change action for gossiping the response was produced. assert_eq!(change_set.len(), 1); @@ -2878,7 +2934,8 @@ pub mod test { ); // 4. ACTION: Our replica attempts to validate the artifact. - let change_set = pool_manager.validate_shares(&canister_http_pool); + let change_set = + pool_manager.validate_shares(&pool_manager.latest_state(), &canister_http_pool); // 5. ASSERTION: The artifact must be invalidated with the specific reason. assert_eq!(change_set.len(), 1); @@ -2986,7 +3043,8 @@ pub mod test { timestamp: UNIX_EPOCH, }); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = pool_manager + .validate_shares(&pool_manager.latest_state(), &canister_http_pool); assert_matches!( &changes[0], @@ -3013,7 +3071,8 @@ pub mod test { timestamp: UNIX_EPOCH, }); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = pool_manager + .validate_shares(&pool_manager.latest_state(), &canister_http_pool); assert_matches!( &changes[0], @@ -3041,7 +3100,8 @@ pub mod test { timestamp: UNIX_EPOCH, }); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = pool_manager + .validate_shares(&pool_manager.latest_state(), &canister_http_pool); assert_matches!( &changes[0], @@ -3068,7 +3128,8 @@ pub mod test { timestamp: UNIX_EPOCH, }); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = pool_manager + .validate_shares(&pool_manager.latest_state(), &canister_http_pool); assert_matches!( &changes[0], @@ -3180,7 +3241,8 @@ pub mod test { timestamp: UNIX_EPOCH, }); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = pool_manager + .validate_shares(&pool_manager.latest_state(), &canister_http_pool); let validation_err = format!( "Response size {} exceeds the maximum allowed size of {}", @@ -3242,7 +3304,8 @@ pub mod test { timestamp: UNIX_EPOCH, }); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = pool_manager + .validate_shares(&pool_manager.latest_state(), &canister_http_pool); assert_matches!(&changes[0], CanisterHttpChangeAction::MoveToValidated(_)); } @@ -3321,7 +3384,8 @@ pub mod test { ); // 3. Call the function and get the change set. - let change_set = pool_manager.create_shares_from_responses(); + let change_set = + pool_manager.create_shares_from_responses(&pool_manager.latest_state()); // 4. Assert that the correct change action for gossiping the response was produced. assert_eq!(change_set.len(), 1); @@ -3431,7 +3495,8 @@ pub mod test { log, ); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = + pool_manager.validate_shares(&pool_manager.latest_state(), &canister_http_pool); assert_eq!(changes.len(), 1); assert_matches!( @@ -3537,7 +3602,8 @@ pub mod test { log, ); - let changes = pool_manager.validate_shares(&canister_http_pool); + let changes = + pool_manager.validate_shares(&pool_manager.latest_state(), &canister_http_pool); // The share must not be invalidated for overspending. assert_eq!(changes.len(), 1); @@ -3675,4 +3741,381 @@ pub mod test { }) }); } + + // =================================================================== + // Asynchronous receipts + // =================================================================== + + /// A share for an outcall that has already been responded to is still validated: + /// it may yet be picked up as an asynchronous receipt. + #[test] + fn test_share_for_delivered_context_is_validated() { + ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| { + with_test_replica_logger(|log| { + let Dependencies { + pool, + replica_config, + crypto, + state_manager, + registry, + .. + } = DependenciesBuilder::new(pool_config.clone(), 5).build(); + + let callback_id = CallbackId::from(0); + state_manager + .get_mut() + .expect_get_latest_state() + .return_const(Labeled::new( + Height::from(1), + Arc::new(state_with_delivered_http_calls(BTreeMap::from([( + callback_id, + test_request_context( + Replication::FullyReplicated, + PricingVersion::PayAsYouGo, + None, + ), + )]))), + )); + + let mut canister_http_pool = + CanisterHttpPoolImpl::new(MetricsRegistry::new(), no_op_logger()); + let receipt_share = CanisterHttpResponseReceipt { + metadata: CanisterHttpResponseMetadata { + id: callback_id, + content_hash: CryptoHashOf::new(CryptoHash(vec![])), + content_size: 0, + is_reject: false, + replica_version: ReplicaVersion::default(), + }, + payment_receipt: CanisterHttpPaymentReceipt::default(), + }; + let signature = crypto + .sign( + &receipt_share, + replica_config.node_id, + RegistryVersion::from(1), + ) + .unwrap(); + canister_http_pool.insert(UnvalidatedArtifact { + message: CanisterHttpResponseArtifact { + share: Signed { + content: receipt_share, + signature, + }, + response: None, + }, + peer_id: replica_config.node_id, + timestamp: UNIX_EPOCH, + }); + + let pool_manager = CanisterHttpPoolManagerImpl::new( + state_manager as Arc<_>, + Arc::new(Mutex::new(Box::new(MockNonBlockingChannel::new()))), + crypto, + pool.get_cache(), + replica_config, + SubnetType::Application, + Arc::clone(®istry) as Arc<_>, + MetricsRegistry::new(), + log, + ); + + let changes = + pool_manager.validate_shares(&pool_manager.latest_state(), &canister_http_pool); + + assert_matches!( + changes.as_slice(), + [CanisterHttpChangeAction::MoveToValidated(share)] + if share.content.id() == callback_id + ); + }) + }); + } + + /// Artifacts of an outcall that has already been responded to are kept for as + /// long as its delivered context is around, and purged once it is gone. + #[test] + fn test_shares_of_delivered_context_are_purged_only_once_it_is_gone() { + for delivered in [true, false] { + ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| { + with_test_replica_logger(|log| { + let Dependencies { + pool, + replica_config, + crypto, + state_manager, + registry, + .. + } = DependenciesBuilder::new(pool_config.clone(), 5).build(); + + let callback_id = CallbackId::from(0); + let context = test_request_context( + Replication::FullyReplicated, + PricingVersion::PayAsYouGo, + None, + ); + let contexts = if delivered { + BTreeMap::from([(callback_id, context.clone())]) + } else { + BTreeMap::new() + }; + let mut state = state_with_delivered_http_calls(contexts); + // Whether or not the context is still around, the callback id must + // have been handed out already. + state + .metadata + .subnet_call_context_manager + .push_context(SubnetCallContext::CanisterHttpRequest(context)); + state + .metadata + .subnet_call_context_manager + .canister_http_request_contexts + .clear(); + state_manager + .get_mut() + .expect_get_latest_state() + .return_const(Labeled::new(Height::from(1), Arc::new(state))); + + let mut canister_http_pool = + CanisterHttpPoolImpl::new(MetricsRegistry::new(), no_op_logger()); + let response = empty_canister_http_response(callback_id.get()); + let receipt_share = CanisterHttpResponseReceipt { + metadata: CanisterHttpResponseMetadata { + id: callback_id, + content_hash: crypto_hash(&response), + content_size: response.content.count_bytes() as u32, + is_reject: false, + replica_version: ReplicaVersion::default(), + }, + payment_receipt: CanisterHttpPaymentReceipt::default(), + }; + let signature = crypto + .sign( + &receipt_share, + replica_config.node_id, + RegistryVersion::from(1), + ) + .unwrap(); + canister_http_pool.apply(vec![CanisterHttpChangeAction::AddToValidated( + Signed { + content: receipt_share, + signature, + }, + response, + )]); + + let pool_manager = CanisterHttpPoolManagerImpl::new( + state_manager as Arc<_>, + Arc::new(Mutex::new(Box::new(MockNonBlockingChannel::new()))), + crypto, + pool.get_cache(), + replica_config, + SubnetType::Application, + Arc::clone(®istry) as Arc<_>, + MetricsRegistry::new(), + log, + ); + + let changes = pool_manager.purge_shares_of_processed_requests( + &pool_manager.latest_state(), + &canister_http_pool, + ); + + if delivered { + assert!(changes.is_empty(), "{changes:?}"); + } else { + assert_matches!( + changes.as_slice(), + [ + CanisterHttpChangeAction::RemoveValidated(_), + CanisterHttpChangeAction::RemoveContent(_), + ] + ); + } + }) + }); + } + } + + /// A share is signed even for an outcall that has already been responded to: the + /// work was done and paid for, so the receipt has to be published for the spend + /// to be settled asynchronously. It is gossiped exactly as it would have been + /// before the response was delivered. + #[test] + fn test_share_is_created_for_a_delivered_context() { + for replication in [ + Replication::FullyReplicated, + Replication::NonReplicated(node_test_id(0)), + Replication::Flexible { + committee: BTreeSet::from([node_test_id(0), node_test_id(1)]), + min_responses: 1, + max_responses: 2, + }, + ] { + ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| { + with_test_replica_logger(|log| { + let Dependencies { + pool, + replica_config, + crypto, + state_manager, + registry, + .. + } = DependenciesBuilder::new(pool_config.clone(), 4).build(); + + let callback_id = CallbackId::from(0); + state_manager + .get_mut() + .expect_get_latest_state() + .return_const(Labeled::new( + Height::from(1), + Arc::new(state_with_delivered_http_calls(BTreeMap::from([( + callback_id, + test_request_context( + replication.clone(), + PricingVersion::PayAsYouGo, + None, + ), + )]))), + )); + + let mut shim_mock = MockNonBlockingChannel::::new(); + let mut sequence = Sequence::new(); + shim_mock + .expect_try_receive() + .times(1) + .returning(move || { + Ok(( + empty_canister_http_response(callback_id.get()), + CanisterHttpPaymentReceipt::default(), + )) + }) + .in_sequence(&mut sequence); + shim_mock + .expect_try_receive() + .times(1) + .returning(|| Err(TryReceiveError::Empty)) + .in_sequence(&mut sequence); + + let pool_manager = CanisterHttpPoolManagerImpl::new( + state_manager, + Arc::new(Mutex::new(Box::new(shim_mock))), + crypto, + pool.get_cache(), + replica_config, + SubnetType::Application, + Arc::clone(®istry) as Arc<_>, + MetricsRegistry::new(), + log, + ); + pool_manager + .requested_id_cache + .borrow_mut() + .insert(callback_id); + + let change_set = + pool_manager.create_shares_from_responses(&pool_manager.latest_state()); + + match replication { + // A peer recomputes the response itself, so only the receipt + // is gossiped. + Replication::FullyReplicated => assert_matches!( + change_set.as_slice(), + [CanisterHttpChangeAction::AddToValidated(share, _)] + if share.content.id() == callback_id + ), + // A peer cannot recompute it, and needs it to validate the + // receipt against. + _ => assert_matches!( + change_set.as_slice(), + [CanisterHttpChangeAction::AddToValidatedAndGossipResponse( + share, + _ + )] if share.content.id() == callback_id + ), + } + // The request is no longer in flight. + assert!( + !pool_manager + .requested_id_cache + .borrow() + .contains(&callback_id) + ); + }); + }); + } + } + + /// No *new* request is made to the HTTP adapter for an outcall that has already + /// been responded to: there is nothing left to do for it, and the allowance of a + /// replica that never started is refunded in full when the delivered context + /// times out. + #[test] + fn test_no_new_request_is_made_for_a_delivered_context() { + ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| { + with_test_replica_logger(|log| { + let Dependencies { + pool, + replica_config, + crypto, + state_manager, + registry, + .. + } = DependenciesBuilder::new(pool_config.clone(), 4).build(); + + let context = test_request_context( + Replication::FullyReplicated, + PricingVersion::PayAsYouGo, + None, + ); + // A state holding one already responded to request and one still + // awaiting a response, so that the assertions below distinguish the + // two rather than just observing an idle pool manager. + let delivered_id = CallbackId::from(0); + let mut state = state_with_delivered_http_calls(BTreeMap::from([( + delivered_id, + context.clone(), + )])); + let active_id = state + .metadata + .subnet_call_context_manager + .push_context(SubnetCallContext::CanisterHttpRequest(context)); + assert_ne!(active_id, delivered_id); + state_manager + .get_mut() + .expect_get_latest_state() + .return_const(Labeled::new(Height::from(1), Arc::new(state))); + + let mut shim_mock = MockNonBlockingChannel::::new(); + #[allow(clippy::result_large_err)] + shim_mock + .expect_send() + .withf(move |request: &CanisterHttpRequest| request.id == active_id) + .times(1) + .returning(|_| Ok(())); + + let pool_manager = CanisterHttpPoolManagerImpl::new( + state_manager as Arc<_>, + Arc::new(Mutex::new(Box::new(shim_mock))), + crypto, + pool.get_cache(), + replica_config, + SubnetType::Application, + Arc::clone(®istry) as Arc<_>, + MetricsRegistry::new(), + log, + ); + + let canister_http_pool = + CanisterHttpPoolImpl::new(MetricsRegistry::new(), no_op_logger()); + pool_manager.make_new_requests(&pool_manager.latest_state(), &canister_http_pool); + + // Only the request that is still awaiting a response was dispatched. + assert_eq!( + *pool_manager.requested_id_cache.borrow(), + BTreeSet::from([active_id]) + ); + }) + }); + } } diff --git a/rs/https_outcalls/consensus/src/test_utils.rs b/rs/https_outcalls/consensus/src/test_utils.rs index b64225752615..d630ffc50d6e 100644 --- a/rs/https_outcalls/consensus/src/test_utils.rs +++ b/rs/https_outcalls/consensus/src/test_utils.rs @@ -39,6 +39,7 @@ impl BatchPayloadBuilder for FakeCanisterHttpPayloadBuilder { out_of_cycles: vec![], flexible_responses: vec![], flexible_errors: vec![], + async_receipts: vec![], }; payload_to_bytes(payload, max_size) } diff --git a/rs/interfaces/src/canister_http.rs b/rs/interfaces/src/canister_http.rs index 54e01b3fa8c7..a0a9fab923a5 100644 --- a/rs/interfaces/src/canister_http.rs +++ b/rs/interfaces/src/canister_http.rs @@ -47,6 +47,20 @@ pub enum InvalidCanisterHttpPayloadReason { }, /// A timeout refers to a CallbackId that is unknown by the StateManager UnknownCallbackId(CallbackId), + /// An asynchronous receipt refers to a CallbackId that the StateManager does not + /// know as an already responded to request, i.e. one that is not among the + /// `delivered_canister_http_request_contexts`. + UnknownDeliveredCallbackId(CallbackId), + /// An asynchronous receipt refers to an already responded to request whose + /// delivered context has timed out, i.e. one that message routing settles and + /// drops in this very block, leaving nothing left to refund. + DeliveredCallbackTimedOut(CallbackId), + /// An asynchronous receipt reports a replica whose spend has already been + /// accounted for, either in the certified state or in a past payload. + AlreadyRefunded { + callback_id: CallbackId, + signer: NodeId, + }, /// A CallbackId was included as a timeout, however the Request has not timed out at all NotTimedOut(CallbackId), /// There was an error with a signature calculation diff --git a/rs/protobuf/def/types/v1/canister_http.proto b/rs/protobuf/def/types/v1/canister_http.proto index bd53b9b3e854..5343fce52f22 100644 --- a/rs/protobuf/def/types/v1/canister_http.proto +++ b/rs/protobuf/def/types/v1/canister_http.proto @@ -135,6 +135,7 @@ message CanisterHttpResponseMessage { FlexibleCanisterHttpResponses flexible_responses = 4; FlexibleCanisterHttpError flexible_error = 5; CanisterHttpOutOfCycles out_of_cycles = 6; + CanisterHttpShare async_receipt = 7; } } diff --git a/rs/protobuf/src/gen/types/types.v1.rs b/rs/protobuf/src/gen/types/types.v1.rs index 8d0d4953a276..166ffe39f1b2 100644 --- a/rs/protobuf/src/gen/types/types.v1.rs +++ b/rs/protobuf/src/gen/types/types.v1.rs @@ -733,7 +733,7 @@ pub mod flexible_canister_http_error { pub struct CanisterHttpResponseMessage { #[prost( oneof = "canister_http_response_message::MessageType", - tags = "1, 2, 3, 4, 5, 6" + tags = "1, 2, 3, 4, 5, 6, 7" )] pub message_type: ::core::option::Option, } @@ -753,6 +753,8 @@ pub mod canister_http_response_message { FlexibleError(super::FlexibleCanisterHttpError), #[prost(message, tag = "6")] OutOfCycles(super::CanisterHttpOutOfCycles), + #[prost(message, tag = "7")] + AsyncReceipt(super::CanisterHttpShare), } } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] diff --git a/rs/types/types/src/batch/canister_http.rs b/rs/types/types/src/batch/canister_http.rs index 6c1fe7ec7f1d..5be3422e593b 100644 --- a/rs/types/types/src/batch/canister_http.rs +++ b/rs/types/types/src/batch/canister_http.rs @@ -35,6 +35,7 @@ pub struct CanisterHttpPayload { pub out_of_cycles: Vec, pub flexible_responses: Vec, pub flexible_errors: Vec, + pub async_receipts: Vec, } /// A fully- or non-replicated HTTP outcall whose committee can no longer cover the @@ -272,6 +273,7 @@ impl CanisterHttpPayload { out_of_cycles, flexible_responses, flexible_errors, + async_receipts, } = self; responses.len() + timeouts.len() @@ -279,6 +281,7 @@ impl CanisterHttpPayload { + out_of_cycles.len() + flexible_responses.len() + flexible_errors.len() + + async_receipts.len() } /// Returns the number of non_timeout responses @@ -290,6 +293,7 @@ impl CanisterHttpPayload { out_of_cycles, flexible_responses, flexible_errors, + async_receipts, } = self; responses.len() + divergence_responses.len() @@ -299,6 +303,7 @@ impl CanisterHttpPayload { .iter() .filter(|error| !matches!(error, FlexibleCanisterHttpError::Timeout { .. })) .count() + + async_receipts.len() } /// Returns true, if this is an empty payload @@ -923,6 +928,7 @@ mod tests { out_of_cycles, flexible_responses, flexible_errors, + async_receipts, timeouts: _, // skipped because there is no dedicated protobuf conversion for this } = payload; @@ -957,6 +963,11 @@ mod tests { let roundtripped = CanisterHttpOutOfCycles::try_from(pb).unwrap(); assert_eq!(error, roundtripped); } + for share in async_receipts { + let pb = pb::CanisterHttpShare::from(share.clone()); + let roundtripped = CanisterHttpResponseShare::try_from(pb).unwrap(); + assert_eq!(share, roundtripped); + } } } }