feat: Further limit message memory on small nodes

rs/config/src/execution_environment.rs

@@ -110,7 +110,53 @@ pub const STOP_CANISTER_TIMEOUT_DURATION: Duration = Duration::from_secs(5 * 60)
 /// potential fragmentation. This limit should be larger than the maximum
 /// canister memory size to guarantee that a message that overwrites the whole
 /// memory can succeed.
-pub(crate) const SUBNET_HEAP_DELTA_CAPACITY: NumBytes = NumBytes::new(140 * GIB);
+pub const SUBNET_HEAP_DELTA_CAPACITY: NumBytes = NumBytes::new(140 * GIB);
+
+/// Message memory is capped at `1 / MESSAGE_MEMORY_HEAP_DELTA_DIVISOR` of the
+/// subnet's heap delta capacity.
+const MESSAGE_MEMORY_HEAP_DELTA_DIVISOR: u64 = 3;
+
+/// The subnet's guaranteed response message memory capacity, capped at
+/// `1 / MESSAGE_MEMORY_HEAP_DELTA_DIVISOR` of `heap_delta_capacity`.
+pub fn guaranteed_response_message_memory_capacity(heap_delta_capacity: NumBytes) -> NumBytes {
+    message_memory_capacity(
+        SUBNET_GUARANTEED_RESPONSE_MESSAGE_MEMORY_CAPACITY,
+        heap_delta_capacity,
+    )
+}
+
+/// The subnet's best-effort message memory capacity, capped at
+/// `1 / MESSAGE_MEMORY_HEAP_DELTA_DIVISOR` of `heap_delta_capacity`.
+pub fn best_effort_message_memory_capacity(heap_delta_capacity: NumBytes) -> NumBytes {
+    message_memory_capacity(
+        SUBNET_BEST_EFFORT_MESSAGE_MEMORY_CAPACITY,
+        heap_delta_capacity,
+    )
+}
+
+/// Test helper for configuring a subnet's `maximum_state_delta`: returns the
+/// heap delta capacity required to allow `message_memory` of message memory,
+/// i.e. the inverse of the cap applied by `message_memory_capacity`. Lives here
+/// so the `MESSAGE_MEMORY_HEAP_DELTA_DIVISOR` factor stays confined to this
+/// module rather than being duplicated across tests.
+pub fn heap_delta_capacity_for_message_memory(message_memory: NumBytes) -> NumBytes {
+    NumBytes::new(message_memory.get() * MESSAGE_MEMORY_HEAP_DELTA_DIVISOR)
+}
+
+/// Caps `configured_default_capacity` at `1 / MESSAGE_MEMORY_HEAP_DELTA_DIVISOR`
+/// of `heap_delta_capacity`.
+///
+/// Message memory is held in RAM alongside the heap delta pages, so this bounds
+/// message memory relative to a subnet's heap delta capacity (which is itself
+/// sized relative to available RAM).
+fn message_memory_capacity(
+    configured_default_capacity: NumBytes,
+    heap_delta_capacity: NumBytes,
+) -> NumBytes {
+    configured_default_capacity.min(NumBytes::new(
+        heap_delta_capacity.get() / MESSAGE_MEMORY_HEAP_DELTA_DIVISOR,
+    ))
+}
 
 /// The maximum number of instructions for inspect_message calls.
 const MAX_INSTRUCTIONS_FOR_MESSAGE_ACCEPTANCE_CALLS: NumInstructions =
@@ -244,14 +290,6 @@ pub struct Config {
     /// the subnet.
     pub subnet_memory_capacity: NumBytes,
 
-    /// The maximum amount of logical storage available to guaranteed response
-    /// canister messages across the whole subnet.
-    pub guaranteed_response_message_memory_capacity: NumBytes,
-
-    /// The maximum amount of logical storage available to best-effort canister
-    /// messages across the whole subnet.
-    pub best_effort_message_memory_capacity: NumBytes,
-
     /// The maximum amount of logical storage available to the ingress history
     /// across the whole subnet.
     pub ingress_history_memory_capacity: NumBytes,
@@ -413,9 +451,6 @@ impl Default for Config {
                 MAX_INSTRUCTIONS_FOR_MESSAGE_ACCEPTANCE_CALLS,
             subnet_memory_threshold: SUBNET_MEMORY_THRESHOLD,
             subnet_memory_capacity: SUBNET_MEMORY_CAPACITY,
-            guaranteed_response_message_memory_capacity:
-                SUBNET_GUARANTEED_RESPONSE_MESSAGE_MEMORY_CAPACITY,
-            best_effort_message_memory_capacity: SUBNET_BEST_EFFORT_MESSAGE_MEMORY_CAPACITY,
             ingress_history_memory_capacity: INGRESS_HISTORY_MEMORY_CAPACITY,
             subnet_wasm_custom_sections_memory_capacity:
                 SUBNET_WASM_CUSTOM_SECTIONS_MEMORY_CAPACITY,

rs/execution_environment/src/execution_environment.rs

@@ -17,7 +17,10 @@ use crate::ic00_permissions::Ic00MethodPermissions;
 use crate::metrics::{CallTreeMetrics, CallTreeMetricsImpl, IngressFilterMetrics};
 use candid::Encode;
 use ic_base_types::PrincipalId;
-use ic_config::execution_environment::Config as ExecutionConfig;
+use ic_config::execution_environment::{
+    Config as ExecutionConfig, SUBNET_HEAP_DELTA_CAPACITY,
+    guaranteed_response_message_memory_capacity,
+};
 use ic_config::flag_status::FlagStatus;
 use ic_crypto_utils_canister_threshold_sig::derive_threshold_public_key;
 use ic_cycles_account_manager::{
@@ -532,8 +535,9 @@ impl ExecutionEnvironment {
             self.subnet_memory_capacity(state.resource_limits()).get() as i64
                 - self.config.subnet_memory_reservation.get() as i64
                 - memory_taken.execution().get() as i64,
-            self.config
-                .guaranteed_response_message_memory_capacity
+            state
+                .metadata
+                .guaranteed_response_message_memory_capacity()
                 .get() as i64
                 - memory_taken.guaranteed_response_messages().get() as i64,
             self.config
@@ -572,8 +576,9 @@ impl ExecutionEnvironment {
         &self,
         state: &ReplicatedState,
     ) -> i64 {
-        self.config
-            .guaranteed_response_message_memory_capacity
+        state
+            .metadata
+            .guaranteed_response_message_memory_capacity()
             .get() as i64
             - state.guaranteed_response_message_memory_taken().get() as i64
     }
@@ -4861,7 +4866,10 @@ pub(crate) fn full_subnet_memory_capacity(
         resource_limits
             .maximum_state_size_or(config.subnet_memory_capacity)
             .get() as i64,
-        config.guaranteed_response_message_memory_capacity.get() as i64,
+        guaranteed_response_message_memory_capacity(
+            resource_limits.maximum_state_delta_or(SUBNET_HEAP_DELTA_CAPACITY),
+        )
+        .get() as i64,
         config.subnet_wasm_custom_sections_memory_capacity.get() as i64,
         NonZeroU64::new(1).expect("scaling_factor must be non zero"),
     )

rs/execution_environment/src/scheduler/test_utilities.rs

@@ -7,6 +7,7 @@ use std::{
 use ic_base_types::{CanisterId, NumBytes, PrincipalId, SubnetId};
 use ic_config::{
     embedders::Config as HypervisorConfig,
+    execution_environment::heap_delta_capacity_for_message_memory,
     flag_status::FlagStatus,
     subnet_config::{SchedulerConfig, SubnetConfig},
 };
@@ -813,7 +814,7 @@ pub(crate) struct SchedulerTestBuilder {
     hypervisor_config: HypervisorConfig,
     initial_canister_cycles: Cycles,
     subnet_memory_capacity: u64,
-    subnet_guaranteed_response_message_memory: u64,
+    maximum_state_delta: Option<NumBytes>,
     subnet_callback_soft_limit: usize,
     canister_guaranteed_callback_quota: usize,
     registry_settings: RegistryExecutionSettings,
@@ -845,9 +846,7 @@ impl Default for SchedulerTestBuilder {
             hypervisor_config,
             initial_canister_cycles: Cycles::new(1_000_000_000_000_000_000),
             subnet_memory_capacity: config.subnet_memory_capacity.get(),
-            subnet_guaranteed_response_message_memory: config
-                .guaranteed_response_message_memory_capacity
-                .get(),
+            maximum_state_delta: None,
             subnet_callback_soft_limit: config.subnet_callback_soft_limit,
             canister_guaranteed_callback_quota: config.canister_guaranteed_callback_quota,
             registry_settings: test_registry_settings(),
@@ -891,7 +890,9 @@ impl SchedulerTestBuilder {
         subnet_guaranteed_response_message_memory: u64,
     ) -> Self {
         Self {
-            subnet_guaranteed_response_message_memory,
+            maximum_state_delta: Some(heap_delta_capacity_for_message_memory(NumBytes::new(
+                subnet_guaranteed_response_message_memory,
+            ))),
             ..self
         }
     }
@@ -981,6 +982,10 @@ impl SchedulerTestBuilder {
         }).unwrap();
 
         let mut state = ReplicatedState::new(self.own_subnet_id, self.subnet_type);
+        {
+            let own_subnet_info = std::sync::Arc::make_mut(&mut state.metadata.own_subnet_info);
+            own_subnet_info.resource_limits.maximum_state_delta = self.maximum_state_delta;
+        }
 
         let mut registry_settings = self.registry_settings;
 
@@ -1060,9 +1065,6 @@ impl SchedulerTestBuilder {
         let config = ic_config::execution_environment::Config {
             allocatable_compute_capacity_in_percent: self.allocatable_compute_capacity_in_percent,
             subnet_memory_capacity: NumBytes::new(self.subnet_memory_capacity),
-            guaranteed_response_message_memory_capacity: NumBytes::new(
-                self.subnet_guaranteed_response_message_memory,
-            ),
             // Keep it simple, no memory reservation for responses.
             subnet_memory_reservation: NumBytes::new(0),
             subnet_callback_soft_limit: self.subnet_callback_soft_limit,

rs/messaging/BUILD.bazel

@@ -110,6 +110,7 @@ rust_ic_test_suite_with_extra_srcs(
         "//rs/protobuf",
         "//rs/registry/keys",
         "//rs/registry/proto_data_provider",
+        "//rs/registry/resource_limits",
         "//rs/registry/routing_table",
         "//rs/registry/subnet_type",
         "//rs/replicated_state",

rs/messaging/src/message_routing.rs

@@ -663,7 +663,6 @@ impl<RegistryClient_: RegistryClient> BatchProcessorImpl<RegistryClient_> {
         )));
         let stream_handler = Box::new(routing::stream_handler::StreamHandlerImpl::new(
             subnet_id,
-            hypervisor_config.clone(),
             metrics_registry,
             &metrics,
             Arc::clone(&time_in_stream_metrics),
@@ -696,7 +695,6 @@ impl<RegistryClient_: RegistryClient> BatchProcessorImpl<RegistryClient_> {
             scheduler,
             demux,
             stream_builder,
-            hypervisor_config.clone(),
             log.clone(),
             metrics.clone(),
         ));

rs/messaging/src/routing/stream_handler.rs

@@ -2,8 +2,6 @@ use crate::message_routing::{
     CRITICAL_ERROR_ILLEGAL_ENGINE_MESSAGE, CRITICAL_ERROR_INDUCT_RESPONSE_FAILED, LatencyMetrics,
     MessageRoutingMetrics,
 };
-use ic_base_types::NumBytes;
-use ic_config::execution_environment::Config as HypervisorConfig;
 use ic_error_types::RejectCode;
 use ic_interfaces::messaging::{
     LABEL_VALUE_CANISTER_METHOD_NOT_FOUND, LABEL_VALUE_CANISTER_NOT_FOUND,
@@ -202,9 +200,6 @@ pub(crate) trait StreamHandler: Send {
 pub(crate) struct StreamHandlerImpl {
     subnet_id: SubnetId,
 
-    /// The memory allocated for guaranteed response messages on the subnet.
-    guaranteed_response_message_memory_capacity: NumBytes,
-
     metrics: StreamHandlerMetrics,
     /// Per-destination-subnet histogram of wall time spent by messages in the
     /// stream before they are garbage collected.
@@ -219,16 +214,13 @@ pub(crate) struct StreamHandlerImpl {
 impl StreamHandlerImpl {
     pub(crate) fn new(
         subnet_id: SubnetId,
-        hypervisor_config: HypervisorConfig,
         metrics_registry: &MetricsRegistry,
         message_routing_metrics: &MessageRoutingMetrics,
         time_in_stream_metrics: Arc<Mutex<LatencyMetrics>>,
         log: ReplicaLogger,
     ) -> Self {
         Self {
             subnet_id,
-            guaranteed_response_message_memory_capacity: hypervisor_config
-                .guaranteed_response_message_memory_capacity,
             metrics: StreamHandlerMetrics::new(metrics_registry, message_routing_metrics),
             time_in_stream_metrics,
             time_in_backlog_metrics: RefCell::new(LatencyMetrics::new_time_in_backlog(
@@ -1258,7 +1250,10 @@ impl StreamHandlerImpl {
     /// difference between the subnet's guaranteed response message memory capacity
     /// and its current usage.
     fn available_guaranteed_response_memory(&self, state: &ReplicatedState) -> i64 {
-        self.guaranteed_response_message_memory_capacity.get() as i64
+        state
+            .metadata
+            .guaranteed_response_message_memory_capacity()
+            .get() as i64
             - state.guaranteed_response_message_memory_taken().get() as i64
     }