diff --git a/rs/messaging/src/routing/stream_builder.rs b/rs/messaging/src/routing/stream_builder.rs index 052afb5f9bf1..46ef0f7e04f2 100644 --- a/rs/messaging/src/routing/stream_builder.rs +++ b/rs/messaging/src/routing/stream_builder.rs @@ -14,7 +14,7 @@ use ic_types::messages::{ MAX_INTER_CANISTER_PAYLOAD_IN_BYTES, MAX_REJECT_MESSAGE_LEN_BYTES, NO_DEADLINE, Payload, RejectContext, Request, RequestOrResponse, Response, StreamMessage, }; -use ic_types::{CountBytes, SubnetId}; +use ic_types::{CanisterId, CountBytes, SubnetId}; use ic_types_cycles::{CompoundCycles, Cycles}; #[cfg(test)] use mockall::automock; @@ -42,6 +42,9 @@ struct StreamBuilderMetrics { pub routed_payload_sizes: Histogram, /// Misrouted messages currently in streams, by remote subnet. pub stream_misrouted_messages: IntGaugeVec, + /// Canister output queues skipped because their destination subnet was + /// cooling down. + pub cooling_down_skipped_queues: IntCounter, /// Critical error for payloads above the maximum supported size. pub critical_error_payload_too_large: IntCounter, /// Critical error for responses dropped due to destination not found. @@ -63,6 +66,7 @@ const METRIC_SIGNALS_END: &str = "mr_signals_end"; const METRIC_ROUTED_MESSAGES: &str = "mr_routed_message_count"; const METRIC_ROUTED_PAYLOAD_SIZES: &str = "mr_routed_payload_size_bytes"; const METRIC_STREAM_MISROUTED_MESSAGES: &str = "mr_stream_misrouted_messages"; +const METRIC_COOLING_DOWN_SKIPPED_QUEUES: &str = "mr_cooling_down_skipped_queues"; const LABEL_TYPE: &str = "type"; const LABEL_STATUS: &str = "status"; @@ -126,6 +130,12 @@ impl StreamBuilderMetrics { "Count of misrouted messages in streams, by remote subnet. Only populated for subnets currently involved in a canister migration.", &[LABEL_REMOTE], ); + let cooling_down_skipped_queues = metrics_registry.int_counter( + METRIC_COOLING_DOWN_SKIPPED_QUEUES, + "Canister output queues skipped because their destination subnet was cooling down. \ + Counted once per queue per round, so the same queue is counted repeatedly for as \ + long as the destination subnet keeps cooling down.", + ); let critical_error_payload_too_large = metrics_registry.error_counter(CRITICAL_ERROR_PAYLOAD_TOO_LARGE); let critical_error_response_destination_not_found = @@ -167,6 +177,7 @@ impl StreamBuilderMetrics { routed_messages, routed_payload_sizes, stream_misrouted_messages, + cooling_down_skipped_queues, critical_error_payload_too_large, critical_error_response_destination_not_found, critical_error_induct_response_failed, @@ -442,6 +453,10 @@ impl StreamBuilderImpl { let refund_limit = self.max_stream_messages / 2; self.route_refunds(&mut state, refund_limit, &network_topology, &mut streams); + // No canister can have the subnet's own principal as its canister ID, so this + // identifies the messages taken from the subnet's own output queues. + let own_subnet_as_canister_id = CanisterId::from(self.subnet_id); + let mut requests_to_reject = Vec::new(); let mut oversized_requests = Vec::new(); let mut engine_requests_to_reject: Vec> = Vec::new(); @@ -458,11 +473,24 @@ impl StreamBuilderImpl { // Cheap to clone, `RequestOrResponse` wraps `Arcs`. let msg = msg.clone(); + let is_from_subnet_queues = msg.sender() == own_subnet_as_canister_id; + match network_topology.route(msg.receiver().get()) { // Destination subnet found. Some(dst_subnet_id) => { - let dst_stream_entry = streams.entry(dst_subnet_id); let is_loopback_stream = self.subnet_id == dst_subnet_id; + + // A cooling down destination subnet must not be sent any messages + // from canister output queues. Retain the message (along with + // everything behind it in the same queue) until the destination stops + // cooling down, rather than rejecting or dropping it. + if !is_from_subnet_queues && network_topology.is_cooling_down(&dst_subnet_id) { + self.metrics.cooling_down_skipped_queues.inc(); + output_iter.exclude_queue(); + continue; + } + + let dst_stream_entry = streams.entry(dst_subnet_id); if !is_loopback_stream && is_at_limit( &dst_stream_entry, diff --git a/rs/messaging/src/routing/stream_builder/tests.rs b/rs/messaging/src/routing/stream_builder/tests.rs index 55f84b615536..28e570d3ba77 100644 --- a/rs/messaging/src/routing/stream_builder/tests.rs +++ b/rs/messaging/src/routing/stream_builder/tests.rs @@ -18,8 +18,8 @@ use ic_replicated_state::{ }; use ic_test_utilities_logger::with_test_replica_logger; use ic_test_utilities_metrics::{ - MetricVec, fetch_histogram_stats, fetch_int_counter_vec, fetch_int_gauge_vec, metric_vec, - nonzero_values, + MetricVec, fetch_histogram_stats, fetch_int_counter, fetch_int_counter_vec, + fetch_int_gauge_vec, metric_vec, nonzero_values, }; use ic_test_utilities_state::new_canister_state; use ic_test_utilities_types::ids::{ @@ -1151,6 +1151,315 @@ fn build_streams_drops_refunds_at_engine_boundary() { } } +/// Tests around subnets that are cooling down. +mod cooling_down { + use super::*; + // Explicit import, to disambiguate from the `std` macro of the same name + // (both are in scope via the glob import above). + use pretty_assertions::assert_eq; + + /// The remote subnet that is cooling down in the tests below. + const COOLING_DOWN_SUBNET: SubnetId = REMOTE_SUBNET; + /// A third subnet, never cooling down, used to check that only the messages to + /// cooling down subnets are held back. + const OTHER_SUBNET: SubnetId = SUBNET_3; + + /// The sender of all messages in the tests below, hosted by `LOCAL_SUBNET`. + const SENDER_CANISTER: CanisterId = CanisterId::from_u64(0); + /// The destination canister, hosted by the cooling down subnet (which is + /// `LOCAL_SUBNET` itself in `new_loopback_cooling_down_fixture()`). + const COOLING_DOWN_CANISTER: CanisterId = CanisterId::from_u64(1); + /// A canister hosted by `OTHER_SUBNET`. + const OTHER_CANISTER: CanisterId = CanisterId::from_u64(2); + + /// The non-zero amount of cycles attached by `cooling_down_message_matrix()`. + const ONE_TRILLION_CYCLES: Cycles = Cycles::new(1_000_000_000_000); + + /// Sets up a fixture with `SENDER_CANISTER` hosted by `LOCAL_SUBNET`, + /// `COOLING_DOWN_CANISTER` by `COOLING_DOWN_SUBNET` (which is cooling down) and + /// `OTHER_CANISTER` by `OTHER_SUBNET` (which is not). + fn new_cooling_down_fixture( + log: &ReplicaLogger, + ) -> (StreamBuilderImpl, ReplicatedState, MetricsRegistry) { + let (stream_builder, mut state, metrics_registry) = new_fixture(log); + + state.metadata.modify_network_topology(|network_topology| { + network_topology.set_subnets(btreemap! { + LOCAL_SUBNET => SubnetTopology::default(), + COOLING_DOWN_SUBNET => SubnetTopology { cooling_down: true, ..Default::default() }, + OTHER_SUBNET => SubnetTopology::default(), + }); + network_topology.set_routing_table( + RoutingTable::try_from(btreemap! { + CanisterIdRange { start: SENDER_CANISTER, end: SENDER_CANISTER } => LOCAL_SUBNET, + CanisterIdRange { start: COOLING_DOWN_CANISTER, end: COOLING_DOWN_CANISTER } => COOLING_DOWN_SUBNET, + CanisterIdRange { start: OTHER_CANISTER, end: OTHER_CANISTER } => OTHER_SUBNET, + }) + .unwrap(), + ); + }); + + (stream_builder, state, metrics_registry) + } + + /// Same as `new_cooling_down_fixture()`, except that `COOLING_DOWN_CANISTER` is + /// hosted by `LOCAL_SUBNET` and it is `LOCAL_SUBNET` that is cooling down; i.e. + /// the messages to `COOLING_DOWN_CANISTER` would go into the loopback stream. + fn new_loopback_cooling_down_fixture( + log: &ReplicaLogger, + ) -> (StreamBuilderImpl, ReplicatedState, MetricsRegistry) { + let (stream_builder, mut state, metrics_registry) = new_fixture(log); + + state.metadata.modify_network_topology(|network_topology| { + network_topology.set_subnets(btreemap! { + LOCAL_SUBNET => SubnetTopology { cooling_down: true, ..Default::default() }, + OTHER_SUBNET => SubnetTopology::default(), + }); + network_topology.set_routing_table( + RoutingTable::try_from(btreemap! { + CanisterIdRange { start: SENDER_CANISTER, end: COOLING_DOWN_CANISTER } => LOCAL_SUBNET, + CanisterIdRange { start: OTHER_CANISTER, end: OTHER_CANISTER } => OTHER_SUBNET, + }) + .unwrap(), + ); + }); + + (stream_builder, state, metrics_registry) + } + + /// Marks `subnet_id` as no longer cooling down. + fn clear_cooling_down(state: &mut ReplicatedState, subnet_id: SubnetId) { + state.metadata.modify_network_topology(|network_topology| { + network_topology + .subnets_mut() + .get_mut(&subnet_id) + .unwrap() + .cooling_down = false; + }); + } + + /// A request from `SENDER_CANISTER` to `receiver`, with the given callback ID. + /// + /// `canister_states_with_outputs()` requires the callback IDs of a canister's + /// requests to match the ones that its `CallContextManager` generates, i.e. to + /// start at 1 and be consecutive in push order. + fn request_from_sender_with_callback( + receiver: CanisterId, + deadline: CoarseTime, + payment: Cycles, + callback_id: u64, + ) -> RequestOrResponse { + RequestBuilder::new() + .sender(SENDER_CANISTER) + .receiver(receiver) + .sender_reply_callback(CallbackId::from(callback_id)) + .deadline(deadline) + .payment(payment) + .build() + .into() + } + + /// A request from `SENDER_CANISTER` to `receiver`, as the sender's first call. + fn request_from_sender( + receiver: CanisterId, + deadline: CoarseTime, + payment: Cycles, + ) -> RequestOrResponse { + request_from_sender_with_callback(receiver, deadline, payment, 1) + } + + /// A response from `SENDER_CANISTER` to `originator`. + fn response_from_sender( + originator: CanisterId, + deadline: CoarseTime, + refund: Cycles, + ) -> RequestOrResponse { + RequestOrResponse::Response(Arc::new(Response { + originator, + respondent: SENDER_CANISTER, + originator_reply_callback: CallbackId::from(1), + refund, + response_payload: Payload::Data(vec![]), + deadline, + })) + } + + /// The matrix of canister messages from `SENDER_CANISTER` to `receiver` covered by + /// the cooling down tests: a request or a response; unbounded-wait or bounded-wait; + /// with no cycles or 1T cycles attached. + /// + /// None of these dimensions makes any difference to a message headed for a cooling + /// down subnet; contrast with `is_illegal_engine_msg` in `build_streams_impl()`. + fn cooling_down_message_matrix( + receiver: CanisterId, + ) -> impl Iterator { + [NO_DEADLINE, SOME_DEADLINE] + .into_iter() + .flat_map(|deadline| { + [Cycles::zero(), ONE_TRILLION_CYCLES] + .into_iter() + .map(move |cycles| (deadline, cycles)) + }) + .flat_map(move |(deadline, cycles)| { + [ + request_from_sender(receiver, deadline, cycles), + response_from_sender(receiver, deadline, cycles), + ] + }) + } + + /// Asserts that no canister message was routed into the stream to `subnet_id`. + fn assert_no_messages_routed(state: &ReplicatedState, subnet_id: SubnetId) { + assert_eq!( + Vec::::new(), + routed_messages(state, subnet_id) + ); + } + + /// Returns the canister messages routed into the stream to `subnet_id`. + fn routed_messages(state: &ReplicatedState, subnet_id: SubnetId) -> Vec { + state + .streams() + .get(&subnet_id) + .map_or(Vec::new(), |stream| { + stream + .messages() + .iter() + .map(|(_, msg)| msg.clone()) + .collect() + }) + } + + /// Returns the raw contents of `sender`'s output queue to `receiver`. + fn output_queue_contents( + state: &ReplicatedState, + sender: CanisterId, + receiver: CanisterId, + ) -> Vec { + state + .canister_state(&sender) + .unwrap() + .system_state + .queues() + .output_queue_iter_for_testing(&receiver) + .into_iter() + .flatten() + .cloned() + .collect() + } + + /// Retrieves the `METRIC_COOLING_DOWN_SKIPPED_QUEUES` counter's value. + fn fetch_cooling_down_skipped_queues(metrics_registry: &MetricsRegistry) -> u64 { + fetch_int_counter(metrics_registry, METRIC_COOLING_DOWN_SKIPPED_QUEUES) + .unwrap_or_else(|| panic!("Counter not found: {METRIC_COOLING_DOWN_SKIPPED_QUEUES}")) + } + + /// Tests that a canister message to a cooling down subnet is retained in the + /// sending canister's output queue -- rather than routed, rejected or dropped -- + /// and that it is routed as soon as the destination subnet stops cooling down. + /// + /// Covers the full matrix of: request vs. response; addressed to a canister hosted + /// by the cooling down subnet vs. to the subnet itself (i.e. its management + /// canister); unbounded-wait vs. bounded-wait; and with no cycles vs. 1T cycles + /// attached. Every combination is exercised twice: with a remote subnet cooling + /// down; and with `LOCAL_SUBNET` itself cooling down, i.e. the loopback stream is + /// not exempt either. + #[test] + fn build_streams_retains_messages_to_cooling_down_subnet() { + for cooling_down_subnet in [COOLING_DOWN_SUBNET, LOCAL_SUBNET] { + // A canister hosted by the cooling down subnet; and the subnet itself, i.e. + // its management canister. + for receiver in [COOLING_DOWN_CANISTER, CanisterId::from(cooling_down_subnet)] { + for msg in cooling_down_message_matrix(receiver) { + with_test_replica_logger(|log| { + let (stream_builder, mut provided_state, metrics_registry) = + if cooling_down_subnet == LOCAL_SUBNET { + new_loopback_cooling_down_fixture(&log) + } else { + new_cooling_down_fixture(&log) + }; + provided_state + .put_canister_states(canister_states_with_outputs(vec![msg.clone()])); + + let mut result_state = stream_builder.build_streams(provided_state); + + // Nothing was routed into the stream to the cooling down subnet; the + // message is still in the sender's output queue; and no reject + // response was generated for it. + assert_no_messages_routed(&result_state, cooling_down_subnet); + assert_eq!( + vec![msg.clone()], + output_queue_contents(&result_state, SENDER_CANISTER, receiver) + ); + assert!( + !result_state + .canister_state(&SENDER_CANISTER) + .unwrap() + .has_input() + ); + + assert_routed_messages_eq(MetricVec::new(), &metrics_registry); + assert_eq!(1, fetch_cooling_down_skipped_queues(&metrics_registry)); + assert_eq_critical_errors(0, 0, 0, &metrics_registry); + + // And it is routed as soon as the subnet stops cooling down. + clear_cooling_down(&mut result_state, cooling_down_subnet); + let result_state = stream_builder.build_streams(result_state); + assert_eq!( + vec![StreamMessage::from(msg)], + routed_messages(&result_state, cooling_down_subnet) + ); + }); + } + } + } + } + + /// Tests that only the messages to the cooling down subnet are held back: a + /// message from the same canister to a canister on a subnet that is not cooling + /// down is routed in the very same round. + #[test] + fn build_streams_routes_messages_to_other_subnets_while_one_is_cooling_down() { + with_test_replica_logger(|log| { + let retained = request_from_sender(COOLING_DOWN_CANISTER, NO_DEADLINE, Cycles::zero()); + // The sender's second call, so it must carry callback ID 2. + let routed = + request_from_sender_with_callback(OTHER_CANISTER, NO_DEADLINE, Cycles::zero(), 2); + + let (stream_builder, mut provided_state, metrics_registry) = + new_cooling_down_fixture(&log); + provided_state.put_canister_states(canister_states_with_outputs(vec![ + retained.clone(), + routed.clone(), + ])); + + let result_state = stream_builder.build_streams(provided_state); + + assert_no_messages_routed(&result_state, COOLING_DOWN_SUBNET); + assert_eq!( + vec![retained], + output_queue_contents(&result_state, SENDER_CANISTER, COOLING_DOWN_CANISTER) + ); + assert_eq!( + vec![StreamMessage::from(routed)], + routed_messages(&result_state, OTHER_SUBNET) + ); + assert_routed_messages_eq( + metric_vec(&[( + &[ + (LABEL_TYPE, LABEL_VALUE_TYPE_REQUEST), + (LABEL_STATUS, LABEL_VALUE_STATUS_SUCCESS), + ], + 1, + )]), + &metrics_registry, + ); + assert_eq!(1, fetch_cooling_down_skipped_queues(&metrics_registry)); + assert_eq_critical_errors(0, 0, 0, &metrics_registry); + }); + } +} + /// Given a stream with some (potentially zero) initial refunds and canister /// messages, tests that `build_streams()` respects the various limits when /// routing additional refunds and canister messages: diff --git a/rs/replicated_state/src/metadata_state.rs b/rs/replicated_state/src/metadata_state.rs index 0100c9dfed6e..6e0ebe7d08ae 100644 --- a/rs/replicated_state/src/metadata_state.rs +++ b/rs/replicated_state/src/metadata_state.rs @@ -427,8 +427,15 @@ pub struct SubnetTopology { pub cost_schedule: CanisterCyclesCostSchedule, pub subnet_admins: BTreeSet, - /// Whether the subnet is "cooling down", i.e. quiescing: it inducts no ingress - /// messages. + /// Whether the subnet is "cooling down", i.e. quiescing. While a subnet is + /// cooling down: + /// + /// * it inducts no ingress messages, so the ingress history becomes free of + /// expiring message statuses; + /// * (on all subnets) no messages are routed to a cooling down subnet -- + /// including into the cooling down subnet's own loopback stream -- but + /// retained in their respective output queues, so that the respective + /// streams can be emptied. pub cooling_down: bool, }