When staging, verify the inboxes. (#4923)

## Motivation

We currently verify the correct order of incoming messages, and their
presence in the inboxes, only when handling a block proposal or
confirmed block, but not when _staging_ a block. This can result in the
client putting an invalid pending block into the wallet. (Thanks
@kikakkz for reporting this!)

## Proposal

Do the check also when staging.

## Test Plan

A regression test was added. (Mostly by Claude.)

## Release Plan

- These changes should be backported to `testnet_conway`, then
    - be released in a new SDK.

## Links

- [reviewer
checklist](https://github.com/linera-io/linera-protocol/blob/main/CONTRIBUTING.md#reviewer-checklist)

Author: afck

linera-core/src/chain_worker/state.rs

@@ -1349,6 +1349,9 @@ where
         let (_, committee) = self.chain.current_committee()?;
         block.check_proposal_size(committee.policy().maximum_block_proposal_size)?;
 
+        self.chain
+            .remove_bundles_from_inboxes(block.timestamp, true, block.incoming_bundles())
+            .await?;
         let executed_block =
             Box::pin(self.execute_block(&block, local_time, round, published_blobs)).await?;
 

linera-core/src/unit_tests/worker_tests.rs

@@ -4229,3 +4229,109 @@ where
 
     Ok(())
 }
+#[test_case(MemoryStorageBuilder::default(); "memory")]
+#[cfg_attr(feature = "rocksdb", test_case(RocksDbStorageBuilder::new().await; "rocks_db"))]
+#[cfg_attr(feature = "dynamodb", test_case(DynamoDbStorageBuilder::default(); "dynamo_db"))]
+#[cfg_attr(feature = "scylladb", test_case(ScyllaDbStorageBuilder::default(); "scylla_db"))]
+#[test_log::test(tokio::test)]
+async fn test_stage_block_with_message_earlier_than_cursor<B>(
+    mut storage_builder: B,
+) -> anyhow::Result<()>
+where
+    B: StorageBuilder,
+{
+    let mut signer = InMemorySigner::new(None);
+    let receiver_public_key = signer.generate_new();
+    let owner = receiver_public_key.into();
+    let mut env = TestEnvironment::new(storage_builder.build().await?, false, false).await;
+    let chain_1_desc = env.add_root_chain(1, owner, Amount::from_tokens(10)).await;
+    let chain_2_desc = env.add_root_chain(2, owner, Amount::ZERO).await;
+    let chain_1 = chain_1_desc.id();
+    let chain_2 = chain_2_desc.id();
+
+    // Simulate a certificate sending two messages from chain_1 to chain_2.
+    let sender_hash = CryptoHash::test_hash("sender block");
+
+    // Process the second message bundle on chain_2. This advances next_cursor_to_remove
+    // to height=0, index=1.
+    let block_proposal = make_first_block(chain_2)
+        .with_incoming_bundle(IncomingBundle {
+            origin: chain_1,
+            bundle: MessageBundle {
+                certificate_hash: sender_hash,
+                height: BlockHeight::ZERO,
+                timestamp: Timestamp::from(0),
+                transaction_index: 1,
+                messages: vec![system_credit_message(Amount::from_tokens(2))
+                    .to_posted(0, MessageKind::Tracked)],
+            },
+            action: MessageAction::Accept,
+        })
+        .into_first_proposal(owner, &signer)
+        .await
+        .unwrap();
+
+    let certificate_chain_2 = env.make_certificate(ConfirmedBlock::new(
+        BlockExecutionOutcome {
+            messages: vec![Vec::new()],
+            previous_message_blocks: BTreeMap::new(),
+            previous_event_blocks: BTreeMap::new(),
+            events: vec![Vec::new()],
+            blobs: vec![Vec::new()],
+            state_hash: SystemExecutionState {
+                balance: Amount::from_tokens(2),
+                ..env.system_execution_state(&chain_2_desc.id())
+            }
+            .into_hash()
+            .await,
+            oracle_responses: vec![Vec::new()],
+            operation_results: vec![],
+        }
+        .with(block_proposal.content.block),
+    ));
+
+    env.worker()
+        .handle_confirmed_certificate(certificate_chain_2.clone(), None)
+        .await?;
+
+    // Now try to stage a block with the earlier message (transaction_index: 0).
+    // This should fail with IncorrectMessageOrder because next_cursor_to_remove
+    // is now at index 1, but we're trying to process index 0.
+    let bad_proposed_block = make_child_block(&certificate_chain_2.into_value())
+        .with_incoming_bundle(IncomingBundle {
+            origin: chain_1,
+            bundle: MessageBundle {
+                certificate_hash: sender_hash,
+                height: BlockHeight::ZERO,
+                timestamp: Timestamp::from(0),
+                transaction_index: 0,
+                messages: vec![
+                    system_credit_message(Amount::ONE).to_posted(0, MessageKind::Tracked)
+                ],
+            },
+            action: MessageAction::Accept,
+        });
+
+    // Test stage_block_execution directly - this should fail with IncorrectMessageOrder.
+    assert_matches!(
+        env.worker()
+            .stage_block_execution(bad_proposed_block.clone(), None, vec![])
+            .await,
+        Err(WorkerError::ChainError(chain_error))
+            if matches!(*chain_error, ChainError::IncorrectMessageOrder { .. })
+    );
+
+    // Also test handle_block_proposal for completeness.
+    let bad_proposal = bad_proposed_block
+        .into_first_proposal(owner, &signer)
+        .await
+        .unwrap();
+
+    assert_matches!(
+        env.worker().handle_block_proposal(bad_proposal).await,
+        Err(WorkerError::ChainError(chain_error))
+            if matches!(*chain_error, ChainError::IncorrectMessageOrder { .. })
+    );
+
+    Ok(())
+}