Merge branch 'ulan/run-659' into 'master'

RUN-659: Add a field for resource reservation to replicated state

This adds a new field for tracking resource reservation to the replicated state.

The field is not used, so this MR is effectively a no-op change. 

See merge request dfinity-lab/public/ic!12722

rs/protobuf/def/state/canister_state_bits/v1/canister_state_bits.proto

@@ -323,4 +323,6 @@ message CanisterStateBits {
   reserved 35;
   repeated ConsumedCyclesByUseCase consumed_cycles_since_replica_started_by_use_cases = 36;
   CanisterHistory canister_history = 37;
+  // Resource reservation cycles.
+  state.queues.v1.Cycles reserved_balance = 38;
 }

rs/protobuf/src/gen/state/state.canister_state_bits.v1.rs

@@ -523,6 +523,9 @@ pub struct CanisterStateBits {
         ::prost::alloc::vec::Vec<ConsumedCyclesByUseCase>,
     #[prost(message, optional, tag = "37")]
     pub canister_history: ::core::option::Option<CanisterHistory>,
+    /// Resource reservation cycles.
+    #[prost(message, optional, tag = "38")]
+    pub reserved_balance: ::core::option::Option<super::super::queues::v1::Cycles>,
     #[prost(oneof = "canister_state_bits::CanisterStatus", tags = "11, 12, 13")]
     pub canister_status: ::core::option::Option<canister_state_bits::CanisterStatus>,
 }

rs/replicated_state/src/canister_state/system_state.rs

@@ -295,6 +295,9 @@ pub struct SystemState {
     /// completes, it will apply `ingress_induction_cycles_debit` to `cycles_balance`.
     ingress_induction_cycles_debit: Cycles,
 
+    /// Resource reservation cycles.
+    reserved_balance: Cycles,
+
     /// Tasks to execute before processing input messages.
     /// Currently the task queue is empty outside of execution rounds.
     pub task_queue: VecDeque<ExecutionTask>,
@@ -675,6 +678,7 @@ impl SystemState {
             queues: CanisterQueues::default(),
             cycles_balance: initial_cycles,
             ingress_induction_cycles_debit: Cycles::zero(),
+            reserved_balance: Cycles::zero(),
             memory_allocation: MemoryAllocation::BestEffort,
             freeze_threshold,
             status,
@@ -715,6 +719,7 @@ impl SystemState {
         canister_metrics: CanisterMetrics,
         cycles_balance: Cycles,
         ingress_induction_cycles_debit: Cycles,
+        reserved_balance: Cycles,
         task_queue: VecDeque<ExecutionTask>,
         global_timer: CanisterTimer,
         canister_version: u64,
@@ -731,6 +736,7 @@ impl SystemState {
             canister_metrics,
             cycles_balance,
             ingress_induction_cycles_debit,
+            reserved_balance,
             task_queue,
             global_timer,
             canister_version,
@@ -759,6 +765,11 @@ impl SystemState {
         self.ingress_induction_cycles_debit
     }
 
+    /// Returns resource reservation cycles.
+    pub fn reserved_balance(&self) -> Cycles {
+        self.reserved_balance
+    }
+
     /// Records the given amount as debit that will be charged from the balance
     /// at some point in the future.
     ///

rs/state_layout/src/state_layout.rs

@@ -138,6 +138,7 @@ pub struct CanisterStateBits {
     pub freeze_threshold: NumSeconds,
     pub cycles_balance: Cycles,
     pub cycles_debit: Cycles,
+    pub reserved_balance: Cycles,
     pub status: CanisterStatus,
     pub scheduled_as_first: u64,
     pub skipped_round_due_to_no_messages: u64,
@@ -1596,6 +1597,7 @@ impl From<CanisterStateBits> for pb_canister_state_bits::CanisterStateBits {
             freeze_threshold: item.freeze_threshold.get(),
             cycles_balance: Some(item.cycles_balance.into()),
             cycles_debit: Some(item.cycles_debit.into()),
+            reserved_balance: Some(item.reserved_balance.into()),
             canister_status: Some((&item.status).into()),
             scheduled_as_first: item.scheduled_as_first,
             skipped_round_due_to_no_messages: item.skipped_round_due_to_no_messages,
@@ -1660,6 +1662,12 @@ impl TryFrom<pb_canister_state_bits::CanisterStateBits> for CanisterStateBits {
             .transpose()?
             .unwrap_or_else(Cycles::zero);
 
+        let reserved_balance = value
+            .reserved_balance
+            .map(|c| c.try_into())
+            .transpose()?
+            .unwrap_or_else(Cycles::zero);
+
         let task_queue = value
             .task_queue
             .into_iter()
@@ -1686,6 +1694,7 @@ impl TryFrom<pb_canister_state_bits::CanisterStateBits> for CanisterStateBits {
             freeze_threshold: NumSeconds::from(value.freeze_threshold),
             cycles_balance,
             cycles_debit,
+            reserved_balance,
             status: try_from_option_field(
                 value.canister_status,
                 "CanisterStateBits::canister_status",

rs/state_layout/src/state_layout/tests.rs

@@ -29,6 +29,7 @@ fn default_canister_state_bits() -> CanisterStateBits {
         freeze_threshold: NumSeconds::from(0),
         cycles_balance: Cycles::zero(),
         cycles_debit: Cycles::zero(),
+        reserved_balance: Cycles::zero(),
         status: CanisterStatus::Stopped,
         scheduled_as_first: 0,
         skipped_round_due_to_no_messages: 0,

rs/state_manager/src/checkpoint.rs

@@ -343,6 +343,7 @@ pub fn load_canister_state<P: ReadPolicy>(
         canister_metrics,
         canister_state_bits.cycles_balance,
         canister_state_bits.cycles_debit,
+        canister_state_bits.reserved_balance,
         canister_state_bits.task_queue.into_iter().collect(),
         CanisterTimer::from_nanos_since_unix_epoch(canister_state_bits.global_timer_nanos),
         canister_state_bits.canister_version,

rs/state_manager/src/tip.rs

@@ -471,6 +471,7 @@ fn serialize_canister_to_tip(
             freeze_threshold: canister_state.system_state.freeze_threshold,
             cycles_balance: canister_state.system_state.balance(),
             cycles_debit: canister_state.system_state.ingress_induction_cycles_debit(),
+            reserved_balance: canister_state.system_state.reserved_balance(),
             execution_state_bits,
             status: canister_state.system_state.status.clone(),
             scheduled_as_first: canister_state