From 35e849628782af5c36ead94aa0ba62cd69f3be63 Mon Sep 17 00:00:00 2001 From: Martin Raszyk Date: Thu, 20 Aug 2026 08:21:57 +0000 Subject: [PATCH 1/6] feat: Do not route subnet messages to a cooling down subnet Messages in the subnet's own output queues (only ever responses, as the management canister makes no calls of its own) are no longer routed into streams to a cooling down subnet, as long as this subnet is not cooling down itself. They are retained in their output queue until the destination subnet stops cooling down, rather than being rejected or dropped; and the skipped queue is counted in `mr_cooling_down_skipped_queues`, exactly like a canister output queue. The subnet's own output queues stay exempt from the source side of the check, so that a cooling down subnet can still respond to the calls it has already accepted: while this subnet is cooling down, its own output responses are routed regardless of whether the destination subnet is cooling down, the loopback stream included. Also renames `is_from_subnet_queues` to `is_subnet_output_response`. Co-Authored-By: Claude Opus 5 (1M context) --- rs/messaging/src/routing/stream_builder.rs | 43 +++-- .../src/routing/stream_builder/tests.rs | 171 ++++++++++++++++++ 2 files changed, 199 insertions(+), 15 deletions(-) diff --git a/rs/messaging/src/routing/stream_builder.rs b/rs/messaging/src/routing/stream_builder.rs index 34d23940661c..d902d3f09bd7 100644 --- a/rs/messaging/src/routing/stream_builder.rs +++ b/rs/messaging/src/routing/stream_builder.rs @@ -42,8 +42,8 @@ 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 this subnet or their destination - /// subnet was cooling down. + /// Output queues skipped because this subnet or 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, @@ -132,8 +132,8 @@ impl StreamBuilderMetrics { ); let cooling_down_skipped_queues = metrics_registry.int_counter( METRIC_COOLING_DOWN_SKIPPED_QUEUES, - "Canister output queues skipped because this subnet or their destination subnet \ - was cooling down. Counted once per queue per round, so the same queue is counted \ + "Output queues skipped because this subnet or their destination subnet was \ + cooling down. Counted once per queue per round, so the same queue is counted \ repeatedly for as long as either subnet keeps cooling down.", ); let critical_error_payload_too_large = @@ -454,7 +454,8 @@ impl StreamBuilderImpl { 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. + // identifies the messages taken from the subnet's own output queues. Those are + // only ever responses, as the management canister makes no calls of its own. let own_subnet_as_canister_id = CanisterId::from(self.subnet_id); let own_subnet_is_cooling_down = network_topology.is_cooling_down(&self.subnet_id); @@ -474,7 +475,7 @@ impl StreamBuilderImpl { // Cheap to clone, `RequestOrResponse` wraps `Arcs`. let msg = msg.clone(); - let is_from_subnet_queues = msg.sender() == own_subnet_as_canister_id; + let is_subnet_output_response = msg.sender() == own_subnet_as_canister_id; match network_topology.route(msg.receiver().get()) { // Destination subnet found. @@ -483,15 +484,27 @@ impl StreamBuilderImpl { let dst_subnet_topology = network_topology.subnets().get(&dst_subnet_id); let dst_subnet_type = dst_subnet_topology.map(|topology| topology.subnet_type); - // No messages from canister output queues are routed while either - // this subnet (the source) or the destination subnet is cooling down; - // not even into the loopback stream. Retain the message (along with - // everything behind it in the same queue) until neither subnet is - // cooling down anymore, rather than rejecting or dropping it. - if !is_from_subnet_queues - && (own_subnet_is_cooling_down - || dst_subnet_topology.is_some_and(|topology| topology.cooling_down)) - { + let dst_subnet_is_cooling_down = + dst_subnet_topology.is_some_and(|topology| topology.cooling_down); + + // No messages from canister output queues are routed while either this + // subnet (the source) or the destination subnet is cooling down; not + // even into the loopback stream. + // + // The subnet's own output responses are exempt from the source side of + // the check, so that a cooling down subnet can still respond to the + // calls it has already accepted. But they are held back just the same + // while the destination subnet is cooling down. + // + // Retain the message (along with everything behind it in the same queue) + // until the subnet that is cooling down stops doing so, rather than + // rejecting or dropping it. + let skip_while_cooling_down = if is_subnet_output_response { + !own_subnet_is_cooling_down && dst_subnet_is_cooling_down + } else { + own_subnet_is_cooling_down || dst_subnet_is_cooling_down + }; + if skip_while_cooling_down { self.metrics.cooling_down_skipped_queues.inc(); output_iter.exclude_queue(); continue; diff --git a/rs/messaging/src/routing/stream_builder/tests.rs b/rs/messaging/src/routing/stream_builder/tests.rs index 20f3984ac111..5c6823c52f40 100644 --- a/rs/messaging/src/routing/stream_builder/tests.rs +++ b/rs/messaging/src/routing/stream_builder/tests.rs @@ -1240,6 +1240,17 @@ mod cooling_down { }); } + /// Marks `subnet_id` as cooling down. + fn mark_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 = true; + }); + } + /// A request from `SENDER_CANISTER` to `receiver`, with the given callback ID. /// /// `canister_states_with_outputs()` requires the callback IDs of a canister's @@ -1310,6 +1321,53 @@ mod cooling_down { }) } + /// The matrix of messages from `LOCAL_SUBNET`'s own output queues to + /// `originator` covered by the cooling down tests: unbounded-wait or + /// bounded-wait; with no cycles or 1T cycles attached. + /// + /// Responses only: the management canister makes no calls of its own, so a + /// request can never be found in the subnet's own output queues. + fn cooling_down_subnet_message_matrix( + originator: CanisterId, + ) -> impl Iterator> { + [NO_DEADLINE, SOME_DEADLINE] + .into_iter() + .flat_map(move |deadline| { + [Cycles::zero(), ONE_TRILLION_CYCLES] + .into_iter() + .map(move |refund| { + Arc::new(Response { + originator, + respondent: CanisterId::from(LOCAL_SUBNET), + originator_reply_callback: CallbackId::from(1), + refund, + response_payload: Payload::Data(vec![]), + deadline, + }) + }) + }) + } + + /// Enqueues `response` into `LOCAL_SUBNET`'s own output queues, first pushing + /// then popping the request it is a response to (which is what reserves the + /// output queue slot). + fn push_subnet_output_response(state: &mut ReplicatedState, response: Arc) { + state + .push_input( + RequestBuilder::new() + .sender(response.originator) + .receiver(response.respondent) + .sender_reply_callback(response.originator_reply_callback) + .build() + .into(), + &mut (i64::MAX / 2), + ) + .unwrap(); + state.pop_subnet_input().unwrap(); + + state.subnet_queues_mut().push_output_response(response); + } + /// 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!( @@ -1350,6 +1408,20 @@ mod cooling_down { .collect() } + /// Returns the raw contents of the subnet's own output queue to `receiver`. + fn subnet_output_queue_contents( + state: &ReplicatedState, + receiver: CanisterId, + ) -> Vec { + state + .subnet_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) @@ -1523,6 +1595,105 @@ mod cooling_down { } } } + + /// Tests that a message from the subnet's own output queues to a cooling down + /// subnet is retained in those queues -- rather than routed, rejected or dropped + /// -- and that it is routed as soon as the destination subnet stops cooling + /// down. I.e. the subnet's own output queues are only exempt from the source + /// side of the check (see + /// `build_streams_routes_subnet_messages_while_cooling_down()`), not from the + /// destination side. + /// + /// Covers the full matrix of: 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. + #[test] + fn build_streams_retains_subnet_messages_to_cooling_down_subnet() { + // A canister hosted by the cooling down subnet; and the subnet itself, i.e. + // its management canister. + for originator in [COOLING_DOWN_CANISTER, CanisterId::from(COOLING_DOWN_SUBNET)] { + for response in cooling_down_subnet_message_matrix(originator) { + with_test_replica_logger(|log| { + let (stream_builder, mut provided_state, metrics_registry) = + new_cooling_down_fixture(&log); + push_subnet_output_response(&mut provided_state, response.clone()); + + let mut result_state = stream_builder.build_streams(provided_state); + + // Nothing was routed into the stream to the cooling down subnet and the + // response is still in the subnet's own output queue. + assert_no_messages_routed(&result_state, COOLING_DOWN_SUBNET); + assert_eq!( + vec![RequestOrResponse::Response(response.clone())], + subnet_output_queue_contents(&result_state, originator) + ); + + 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 destination 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(RequestOrResponse::Response(response))], + routed_messages(&result_state, COOLING_DOWN_SUBNET) + ); + }); + } + } + } + + /// Tests that the subnet's own output queues are exempt from the source side of + /// the check: a message from `LOCAL_SUBNET`'s own output queues is routed even + /// while `LOCAL_SUBNET` itself is cooling down, so that a cooling down subnet + /// can still respond to the calls it has already accepted. This holds regardless + /// of whether the destination subnet is cooling down as well, the loopback + /// stream included. + #[test] + fn build_streams_routes_subnet_messages_while_cooling_down() { + for (originator, dst_subnet) in [ + // Destination subnet not cooling down: a canister hosted by `OTHER_SUBNET`; + // and `OTHER_SUBNET` itself, i.e. its management canister. + (OTHER_CANISTER, OTHER_SUBNET), + (CanisterId::from(OTHER_SUBNET), OTHER_SUBNET), + // Destination subnet cooling down as well: a canister hosted by the remote + // `COOLING_DOWN_SUBNET`; and a canister hosted by `LOCAL_SUBNET` itself, + // i.e. the loopback stream. + (COOLING_DOWN_CANISTER, COOLING_DOWN_SUBNET), + (SENDER_CANISTER, LOCAL_SUBNET), + ] { + for response in cooling_down_subnet_message_matrix(originator) { + with_test_replica_logger(|log| { + let (stream_builder, mut provided_state, metrics_registry) = + new_cooling_down_fixture(&log); + mark_cooling_down(&mut provided_state, LOCAL_SUBNET); + push_subnet_output_response(&mut provided_state, response.clone()); + + let result_state = stream_builder.build_streams(provided_state); + + assert_eq!( + vec![StreamMessage::from(RequestOrResponse::Response(response))], + routed_messages(&result_state, dst_subnet) + ); + assert!(subnet_output_queue_contents(&result_state, originator).is_empty()); + + assert_routed_messages_eq( + metric_vec(&[( + &[ + (LABEL_TYPE, LABEL_VALUE_TYPE_RESPONSE), + (LABEL_STATUS, LABEL_VALUE_STATUS_SUCCESS), + ], + 1, + )]), + &metrics_registry, + ); + assert_eq!(0, 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 From 58c0efc6246fcd9b1ea3960b8e3308cc33c3ad29 Mon Sep 17 00:00:00 2001 From: Martin Raszyk Date: Thu, 20 Aug 2026 10:01:19 +0000 Subject: [PATCH 2/6] chore: Fix comment and simplify tests for subnet output responses Corrects the comment on the cooling down check: the subnet's own output responses are exempt for as long as this subnet is cooling down, whether or not the destination subnet is cooling down; they are only held back if this subnet is not cooling down but the destination subnet is. The previous wording claimed they were held back whenever the destination subnet was cooling down, which does not hold. On the test side: drops the `mark_cooling_down()` helper in favor of the existing `new_local_cooling_down_fixture()`; flattens `cooling_down_subnet_message_matrix()` into a list of `(deadline, refund)` pairs; and drops the cases with a management canister as the response originator, as the management canister makes no calls of its own and can therefore never be one. Co-Authored-By: Claude Opus 5 (1M context) --- rs/messaging/src/routing/stream_builder.rs | 11 +- .../src/routing/stream_builder/tests.rs | 142 ++++++++---------- 2 files changed, 66 insertions(+), 87 deletions(-) diff --git a/rs/messaging/src/routing/stream_builder.rs b/rs/messaging/src/routing/stream_builder.rs index d902d3f09bd7..16655daf0884 100644 --- a/rs/messaging/src/routing/stream_builder.rs +++ b/rs/messaging/src/routing/stream_builder.rs @@ -491,13 +491,14 @@ impl StreamBuilderImpl { // subnet (the source) or the destination subnet is cooling down; not // even into the loopback stream. // - // The subnet's own output responses are exempt from the source side of - // the check, so that a cooling down subnet can still respond to the - // calls it has already accepted. But they are held back just the same - // while the destination subnet is cooling down. + // The subnet's own output responses are exempt for as long as this + // subnet is cooling down, so that it can still respond to the calls it + // has already accepted; whether or not the destination subnet is cooling + // down, the loopback stream included. They are only held back if this + // subnet is not cooling down but the destination subnet is. // // Retain the message (along with everything behind it in the same queue) - // until the subnet that is cooling down stops doing so, rather than + // until the subnet that holds it back stops cooling down, rather than // rejecting or dropping it. let skip_while_cooling_down = if is_subnet_output_response { !own_subnet_is_cooling_down && dst_subnet_is_cooling_down diff --git a/rs/messaging/src/routing/stream_builder/tests.rs b/rs/messaging/src/routing/stream_builder/tests.rs index 5c6823c52f40..2b98473a1853 100644 --- a/rs/messaging/src/routing/stream_builder/tests.rs +++ b/rs/messaging/src/routing/stream_builder/tests.rs @@ -1240,17 +1240,6 @@ mod cooling_down { }); } - /// Marks `subnet_id` as cooling down. - fn mark_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 = true; - }); - } - /// A request from `SENDER_CANISTER` to `receiver`, with the given callback ID. /// /// `canister_states_with_outputs()` requires the callback IDs of a canister's @@ -1330,22 +1319,23 @@ mod cooling_down { fn cooling_down_subnet_message_matrix( originator: CanisterId, ) -> impl Iterator> { - [NO_DEADLINE, SOME_DEADLINE] - .into_iter() - .flat_map(move |deadline| { - [Cycles::zero(), ONE_TRILLION_CYCLES] - .into_iter() - .map(move |refund| { - Arc::new(Response { - originator, - respondent: CanisterId::from(LOCAL_SUBNET), - originator_reply_callback: CallbackId::from(1), - refund, - response_payload: Payload::Data(vec![]), - deadline, - }) - }) + [ + (NO_DEADLINE, Cycles::zero()), + (NO_DEADLINE, ONE_TRILLION_CYCLES), + (SOME_DEADLINE, Cycles::zero()), + (SOME_DEADLINE, ONE_TRILLION_CYCLES), + ] + .into_iter() + .map(move |(deadline, refund)| { + Arc::new(Response { + originator, + respondent: CanisterId::from(LOCAL_SUBNET), + originator_reply_callback: CallbackId::from(1), + refund, + response_payload: Payload::Data(vec![]), + deadline, }) + }) } /// Enqueues `response` into `LOCAL_SUBNET`'s own output queues, first pushing @@ -1596,78 +1586,66 @@ mod cooling_down { } } - /// Tests that a message from the subnet's own output queues to a cooling down - /// subnet is retained in those queues -- rather than routed, rejected or dropped - /// -- and that it is routed as soon as the destination subnet stops cooling - /// down. I.e. the subnet's own output queues are only exempt from the source - /// side of the check (see - /// `build_streams_routes_subnet_messages_while_cooling_down()`), not from the - /// destination side. + /// Tests that a response in the subnet's own output queues addressed to a + /// canister on a cooling down subnet is retained there -- rather than routed, + /// rejected or dropped -- while `LOCAL_SUBNET` is not cooling down; and that it + /// is routed as soon as the destination subnet stops cooling down. + /// + /// Covers the full matrix of: unbounded-wait vs. bounded-wait; and with no + /// cycles vs. 1T cycles attached. /// - /// Covers the full matrix of: 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. + /// Contrast with `build_streams_routes_subnet_messages_while_cooling_down()`, + /// where `LOCAL_SUBNET` is cooling down and the response is routed regardless. #[test] fn build_streams_retains_subnet_messages_to_cooling_down_subnet() { - // A canister hosted by the cooling down subnet; and the subnet itself, i.e. - // its management canister. - for originator in [COOLING_DOWN_CANISTER, CanisterId::from(COOLING_DOWN_SUBNET)] { - for response in cooling_down_subnet_message_matrix(originator) { - with_test_replica_logger(|log| { - let (stream_builder, mut provided_state, metrics_registry) = - new_cooling_down_fixture(&log); - push_subnet_output_response(&mut provided_state, response.clone()); - - let mut result_state = stream_builder.build_streams(provided_state); - - // Nothing was routed into the stream to the cooling down subnet and the - // response is still in the subnet's own output queue. - assert_no_messages_routed(&result_state, COOLING_DOWN_SUBNET); - assert_eq!( - vec![RequestOrResponse::Response(response.clone())], - subnet_output_queue_contents(&result_state, originator) - ); + for response in cooling_down_subnet_message_matrix(COOLING_DOWN_CANISTER) { + with_test_replica_logger(|log| { + let (stream_builder, mut provided_state, metrics_registry) = + new_cooling_down_fixture(&log); + push_subnet_output_response(&mut provided_state, response.clone()); + + let mut result_state = stream_builder.build_streams(provided_state); + + // Nothing was routed into the stream to the cooling down subnet and the + // response is still in the subnet's own output queue. + assert_no_messages_routed(&result_state, COOLING_DOWN_SUBNET); + assert_eq!( + vec![RequestOrResponse::Response(response.clone())], + subnet_output_queue_contents(&result_state, COOLING_DOWN_CANISTER) + ); - 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); + 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 destination 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(RequestOrResponse::Response(response))], - routed_messages(&result_state, COOLING_DOWN_SUBNET) - ); - }); - } + // And it is routed as soon as the destination 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(RequestOrResponse::Response(response))], + routed_messages(&result_state, COOLING_DOWN_SUBNET) + ); + }); } } - /// Tests that the subnet's own output queues are exempt from the source side of - /// the check: a message from `LOCAL_SUBNET`'s own output queues is routed even - /// while `LOCAL_SUBNET` itself is cooling down, so that a cooling down subnet - /// can still respond to the calls it has already accepted. This holds regardless - /// of whether the destination subnet is cooling down as well, the loopback - /// stream included. + /// Tests that a response in `LOCAL_SUBNET`'s own output queues is routed while + /// `LOCAL_SUBNET` itself is cooling down -- so that a cooling down subnet can + /// still respond to the calls it has already accepted -- whether or not the + /// destination subnet is cooling down, the loopback stream included. #[test] fn build_streams_routes_subnet_messages_while_cooling_down() { for (originator, dst_subnet) in [ - // Destination subnet not cooling down: a canister hosted by `OTHER_SUBNET`; - // and `OTHER_SUBNET` itself, i.e. its management canister. + // A canister hosted by `OTHER_SUBNET`, which is not cooling down. (OTHER_CANISTER, OTHER_SUBNET), - (CanisterId::from(OTHER_SUBNET), OTHER_SUBNET), - // Destination subnet cooling down as well: a canister hosted by the remote - // `COOLING_DOWN_SUBNET`; and a canister hosted by `LOCAL_SUBNET` itself, - // i.e. the loopback stream. - (COOLING_DOWN_CANISTER, COOLING_DOWN_SUBNET), - (SENDER_CANISTER, LOCAL_SUBNET), + // A canister hosted by `LOCAL_SUBNET`, i.e. the loopback stream; and + // `LOCAL_SUBNET` is cooling down. + (COOLING_DOWN_CANISTER, LOCAL_SUBNET), ] { for response in cooling_down_subnet_message_matrix(originator) { with_test_replica_logger(|log| { let (stream_builder, mut provided_state, metrics_registry) = - new_cooling_down_fixture(&log); - mark_cooling_down(&mut provided_state, LOCAL_SUBNET); + new_local_cooling_down_fixture(&log); push_subnet_output_response(&mut provided_state, response.clone()); let result_state = stream_builder.build_streams(provided_state); From ae62a2bb9a951d3c6c30bcddce757b9229a081d5 Mon Sep 17 00:00:00 2001 From: Martin Raszyk Date: Thu, 20 Aug 2026 10:19:02 +0000 Subject: [PATCH 3/6] chore: Separate `own_subnet_is_cooling_down` from the line above `own_subnet_as_canister_id` is explained by the comment right above it, which does not apply to `own_subnet_is_cooling_down`. Co-Authored-By: Claude Opus 5 (1M context) --- rs/messaging/src/routing/stream_builder.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/rs/messaging/src/routing/stream_builder.rs b/rs/messaging/src/routing/stream_builder.rs index 16655daf0884..28f6ed9f6de2 100644 --- a/rs/messaging/src/routing/stream_builder.rs +++ b/rs/messaging/src/routing/stream_builder.rs @@ -457,6 +457,7 @@ impl StreamBuilderImpl { // identifies the messages taken from the subnet's own output queues. Those are // only ever responses, as the management canister makes no calls of its own. let own_subnet_as_canister_id = CanisterId::from(self.subnet_id); + let own_subnet_is_cooling_down = network_topology.is_cooling_down(&self.subnet_id); let mut requests_to_reject = Vec::new(); From c2e77f1eaee99106d0ae7b2107e83780b65803cc Mon Sep 17 00:00:00 2001 From: Martin Raszyk Date: Thu, 20 Aug 2026 10:26:19 +0000 Subject: [PATCH 4/6] test: Cover a cooling down remote destination subnet `build_streams_routes_subnet_messages_while_cooling_down()` only covered a destination subnet that was not cooling down and the loopback stream. Restore the `mark_cooling_down()` helper and switch back to `new_cooling_down_fixture()` so that a response addressed to a canister on a remote cooling down subnet is covered as well: it is routed all the same, as `LOCAL_SUBNET` is cooling down. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/routing/stream_builder/tests.rs | 25 +++++++++++++++---- 1 file changed, 20 insertions(+), 5 deletions(-) diff --git a/rs/messaging/src/routing/stream_builder/tests.rs b/rs/messaging/src/routing/stream_builder/tests.rs index 2b98473a1853..d6686f914582 100644 --- a/rs/messaging/src/routing/stream_builder/tests.rs +++ b/rs/messaging/src/routing/stream_builder/tests.rs @@ -1164,7 +1164,8 @@ mod cooling_down { /// 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`. + /// The sender of all canister 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_local_cooling_down_fixture()`). @@ -1240,6 +1241,17 @@ mod cooling_down { }); } + /// Marks `subnet_id` as cooling down. + fn mark_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 = true; + }); + } + /// A request from `SENDER_CANISTER` to `receiver`, with the given callback ID. /// /// `canister_states_with_outputs()` requires the callback IDs of a canister's @@ -1638,14 +1650,17 @@ mod cooling_down { for (originator, dst_subnet) in [ // A canister hosted by `OTHER_SUBNET`, which is not cooling down. (OTHER_CANISTER, OTHER_SUBNET), - // A canister hosted by `LOCAL_SUBNET`, i.e. the loopback stream; and - // `LOCAL_SUBNET` is cooling down. - (COOLING_DOWN_CANISTER, LOCAL_SUBNET), + // A canister hosted by the remote `COOLING_DOWN_SUBNET`. + (COOLING_DOWN_CANISTER, COOLING_DOWN_SUBNET), + // A canister hosted by `LOCAL_SUBNET` itself, i.e. the loopback stream; + // and `LOCAL_SUBNET` is cooling down. + (SENDER_CANISTER, LOCAL_SUBNET), ] { for response in cooling_down_subnet_message_matrix(originator) { with_test_replica_logger(|log| { let (stream_builder, mut provided_state, metrics_registry) = - new_local_cooling_down_fixture(&log); + new_cooling_down_fixture(&log); + mark_cooling_down(&mut provided_state, LOCAL_SUBNET); push_subnet_output_response(&mut provided_state, response.clone()); let result_state = stream_builder.build_streams(provided_state); From 161581ba4d9c31876b2890f379b68f6c507c4af4 Mon Sep 17 00:00:00 2001 From: mraszyk <31483726+mraszyk@users.noreply.github.com> Date: Thu, 20 Aug 2026 12:46:56 +0200 Subject: [PATCH 5/6] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- rs/messaging/src/routing/stream_builder/tests.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/rs/messaging/src/routing/stream_builder/tests.rs b/rs/messaging/src/routing/stream_builder/tests.rs index d6686f914582..ff3f2778607f 100644 --- a/rs/messaging/src/routing/stream_builder/tests.rs +++ b/rs/messaging/src/routing/stream_builder/tests.rs @@ -1356,10 +1356,12 @@ mod cooling_down { fn push_subnet_output_response(state: &mut ReplicatedState, response: Arc) { state .push_input( - RequestBuilder::new() +RequestBuilder::new() .sender(response.originator) .receiver(response.respondent) .sender_reply_callback(response.originator_reply_callback) + .payment(response.refund) + .deadline(response.deadline) .build() .into(), &mut (i64::MAX / 2), From b84c239d597a5638523d9c17a757b00bdd48fc0e Mon Sep 17 00:00:00 2001 From: IDX GitHub Automation Date: Thu, 20 Aug 2026 10:52:50 +0000 Subject: [PATCH 6/6] Automatically fixing code for linting and formatting issues --- rs/messaging/src/routing/stream_builder/tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rs/messaging/src/routing/stream_builder/tests.rs b/rs/messaging/src/routing/stream_builder/tests.rs index ff3f2778607f..6d870c91d2ca 100644 --- a/rs/messaging/src/routing/stream_builder/tests.rs +++ b/rs/messaging/src/routing/stream_builder/tests.rs @@ -1356,7 +1356,7 @@ mod cooling_down { fn push_subnet_output_response(state: &mut ReplicatedState, response: Arc) { state .push_input( -RequestBuilder::new() + RequestBuilder::new() .sender(response.originator) .receiver(response.respondent) .sender_reply_callback(response.originator_reply_callback)