Merge branch 'alin/validate-task-queue-length' into 'master'

fix: Validate that the task queue has at most one entry between rounds

Between execution rounds (and when loading a checkpoint) there should be no
`Heartbeat` or `Timer` tasks in a canister's task queue, only paused/aborted
executions/installs (already checked / ensured).

And there should be at most one of those (check added in this MR). 

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

rs/execution_environment/src/scheduler.rs

@@ -1387,6 +1387,13 @@ impl SchedulerImpl {
                     }
                 }
             }
+
+            // There should be at most one paused or aborted task left in the task queue.
+            assert!(
+                canister.system_state.task_queue.len() <= 1,
+                "Unexpected tasks left in the task queue of canister {} after a round in canister {:?}",
+                id, canister.system_state.task_queue
+            );
         }
     }
 }

rs/state_layout/src/state_layout.rs

@@ -1951,11 +1951,17 @@ impl TryFrom<pb_canister_state_bits::CanisterStateBits> for CanisterStateBits {
             .map(|c| c.into())
             .unwrap_or_else(Cycles::zero);
 
-        let task_queue = value
+        let task_queue: Vec<_> = value
             .task_queue
             .into_iter()
             .map(|v| v.try_into())
             .collect::<Result<_, _>>()?;
+        if task_queue.len() > 1 {
+            return Err(ProxyDecodeError::Other(format!(
+                "Expecting at most one task queue entry. Found {:?}",
+                task_queue
+            )));
+        }
 
         let mut consumed_cycles_since_replica_started_by_use_cases = BTreeMap::new();
         for x in value

rs/state_layout/src/state_layout/tests.rs

@@ -225,7 +225,7 @@ fn test_encode_decode_task_queue() {
             .respondent(canister_test_id(42))
             .build(),
     );
-    let task_queue = vec![
+    for task in [
         ExecutionTask::AbortedInstallCode {
             message: CanisterCall::Ingress(Arc::clone(&ingress)),
             prepaid_execution_cycles: Cycles::new(1),
@@ -248,15 +248,17 @@ fn test_encode_decode_task_queue() {
             input: CanisterMessageOrTask::Message(CanisterMessage::Ingress(Arc::clone(&ingress))),
             prepaid_execution_cycles: Cycles::new(5),
         },
-    ];
-    let canister_state_bits = CanisterStateBits {
-        task_queue: task_queue.clone(),
-        ..default_canister_state_bits()
-    };
-
-    let pb_bits = pb_canister_state_bits::CanisterStateBits::from(canister_state_bits);
-    let canister_state_bits = CanisterStateBits::try_from(pb_bits).unwrap();
-    assert_eq!(canister_state_bits.task_queue, task_queue);
+    ] {
+        let task_queue = vec![task];
+        let canister_state_bits = CanisterStateBits {
+            task_queue: task_queue.clone(),
+            ..default_canister_state_bits()
+        };
+
+        let pb_bits = pb_canister_state_bits::CanisterStateBits::from(canister_state_bits);
+        let canister_state_bits = CanisterStateBits::try_from(pb_bits).unwrap();
+        assert_eq!(canister_state_bits.task_queue, task_queue);
+    }
 }
 
 #[test]