Process epoch changes and events every time a block is executed (#4799)

## Motivation

We would only process epoch changes and event streams when explicitly
processing the inbox, which means it was possible eg. for users to
unintentionally skip epoch updates while keeping proposing blocks on
their chains.

## Proposal

Process epoch changes (and any other relevant event streams) every time
a block is proposed.

## Test Plan

CI

## Release Plan

- These changes should be backported to the latest `testnet` branch,
then
    - be released in a new SDK,

## Links

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

Author: bart-linera

linera-chain/src/data_types/mod.rs

@@ -159,6 +159,15 @@ pub enum Transaction {
 
 impl BcsHashable<'_> for Transaction {}
 
+impl Transaction {
+    pub fn incoming_bundle(&self) -> Option<&IncomingBundle> {
+        match self {
+            Transaction::ReceiveMessages(bundle) => Some(bundle),
+            _ => None,
+        }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, SimpleObject)]
 #[graphql(name = "Operation")]
 pub struct OperationMetadata {

linera-core/src/client/mod.rs

@@ -2122,75 +2122,6 @@ impl<Env: Environment> ChainClient<Env> {
         Ok(info)
     }
 
-    /// Submits a fast block proposal to the validators.
-    ///
-    /// This must only be used with valid epoch and super owner.
-    #[instrument(level = "trace", skip(committee, operations))]
-    pub async fn submit_fast_block_proposal(
-        &self,
-        committee: &Committee,
-        operations: &[Operation],
-        incoming_bundles: &[IncomingBundle],
-        super_owner: AccountOwner,
-    ) -> Result<(u64, u64, u64, u64), ChainClientError> {
-        let creating_proposal_start = Instant::now();
-        let info = self.chain_info().await?;
-        let timestamp = self.next_timestamp(incoming_bundles, info.timestamp);
-        let transactions = incoming_bundles
-            .iter()
-            .map(|bundle| Transaction::ReceiveMessages(bundle.clone()))
-            .chain(
-                operations
-                    .iter()
-                    .map(|operation| Transaction::ExecuteOperation(operation.clone())),
-            )
-            .collect::<Vec<_>>();
-        let proposed_block = ProposedBlock {
-            epoch: info.epoch,
-            chain_id: self.chain_id,
-            transactions,
-            previous_block_hash: info.block_hash,
-            height: info.next_block_height,
-            authenticated_owner: Some(super_owner),
-            timestamp,
-        };
-        let proposal = Box::new(
-            BlockProposal::new_initial(
-                super_owner,
-                Round::Fast,
-                proposed_block.clone(),
-                self.signer(),
-            )
-            .await
-            .map_err(ChainClientError::signer_failure)?,
-        );
-        let creating_proposal_ms = creating_proposal_start.elapsed().as_millis() as u64;
-        let stage_block_execution_start = Instant::now();
-        let block = self
-            .client
-            .local_node
-            .stage_block_execution(proposed_block, None, Vec::new())
-            .await?
-            .0;
-        let stage_block_execution_ms = stage_block_execution_start.elapsed().as_millis() as u64;
-        let creating_confirmed_block_start = Instant::now();
-        let value = ConfirmedBlock::new(block);
-        let creating_confirmed_block_ms =
-            creating_confirmed_block_start.elapsed().as_millis() as u64;
-        let submitting_block_proposal_start = Instant::now();
-        self.client
-            .submit_block_proposal(committee, proposal, value)
-            .await?;
-        let submitting_block_proposal_ms =
-            submitting_block_proposal_start.elapsed().as_millis() as u64;
-        Ok((
-            creating_proposal_ms,
-            stage_block_execution_ms,
-            creating_confirmed_block_ms,
-            submitting_block_proposal_ms,
-        ))
-    }
-
     /// Attempts to update all validators about the local chain.
     #[instrument(level = "trace", skip(old_committee))]
     pub async fn update_validators(
@@ -2706,6 +2637,26 @@ impl<Env: Environment> ChainClient<Env> {
         &self,
         operations: Vec<Operation>,
         blobs: Vec<Blob>,
+    ) -> Result<ExecuteBlockOutcome, ChainClientError> {
+        let transactions = self.prepend_epochs_messages_and_events(operations).await?;
+
+        if transactions.is_empty() {
+            return Err(ChainClientError::LocalNodeError(
+                LocalNodeError::WorkerError(WorkerError::ChainError(Box::new(
+                    ChainError::EmptyBlock,
+                ))),
+            ));
+        }
+
+        self.execute_prepared_transactions(transactions, blobs)
+            .await
+    }
+
+    #[instrument(level = "trace", skip(transactions, blobs))]
+    async fn execute_prepared_transactions(
+        &self,
+        transactions: Vec<Transaction>,
+        blobs: Vec<Blob>,
     ) -> Result<ExecuteBlockOutcome, ChainClientError> {
         #[cfg(with_metrics)]
         let _latency = metrics::EXECUTE_BLOCK_LATENCY.measure_latency();
@@ -2723,16 +2674,10 @@ impl<Env: Environment> ChainClient<Env> {
             ClientOutcome::Committed(None) => {}
         }
 
-        let incoming_bundles = self.pending_message_bundles().await?;
-        let identity = self.identity().await?;
-        let confirmed_value = self
-            .new_pending_block(incoming_bundles, operations, blobs, identity)
-            .await?;
+        let block = self.new_pending_block(transactions, blobs).await?;
 
         match self.process_pending_block_without_prepare().await? {
-            ClientOutcome::Committed(Some(certificate))
-                if certificate.block() == confirmed_value.block() =>
-            {
+            ClientOutcome::Committed(Some(certificate)) if certificate.block() == &block => {
                 Ok(ExecuteBlockOutcome::Executed(certificate))
             }
             ClientOutcome::Committed(Some(certificate)) => {
@@ -2748,17 +2693,48 @@ impl<Env: Environment> ChainClient<Env> {
         }
     }
 
+    /// Creates a vector of transactions which, in addition to the provided operations,
+    /// also contains epoch changes, receiving message bundles and event stream updates
+    /// (if there are any to be processed).
+    /// This should be called when executing a block, in order to make sure that any pending
+    /// messages or events are included in it.
+    #[instrument(level = "trace", skip(operations))]
+    async fn prepend_epochs_messages_and_events(
+        &self,
+        operations: Vec<Operation>,
+    ) -> Result<Vec<Transaction>, ChainClientError> {
+        let incoming_bundles = self.pending_message_bundles().await?;
+        let stream_updates = self.collect_stream_updates().await?;
+        Ok(self
+            .collect_epoch_changes()
+            .await?
+            .into_iter()
+            .map(Transaction::ExecuteOperation)
+            .chain(
+                incoming_bundles
+                    .into_iter()
+                    .map(Transaction::ReceiveMessages),
+            )
+            .chain(
+                stream_updates
+                    .into_iter()
+                    .map(Transaction::ExecuteOperation),
+            )
+            .chain(operations.into_iter().map(Transaction::ExecuteOperation))
+            .collect::<Vec<_>>())
+    }
+
     /// Creates a new pending block and handles the proposal in the local node.
     /// Next time `process_pending_block_without_prepare` is called, this block will be proposed
     /// to the validators.
-    #[instrument(level = "trace", skip(incoming_bundles, operations, blobs))]
+    #[instrument(level = "trace", skip(transactions, blobs))]
     async fn new_pending_block(
         &self,
-        incoming_bundles: Vec<IncomingBundle>,
-        operations: Vec<Operation>,
+        transactions: Vec<Transaction>,
         blobs: Vec<Blob>,
-        identity: AccountOwner,
-    ) -> Result<ConfirmedBlock, ChainClientError> {
+    ) -> Result<Block, ChainClientError> {
+        let identity = self.identity().await?;
+
         ensure!(
             self.pending_proposal().is_none(),
             ChainClientError::BlockProposalError(
@@ -2767,12 +2743,7 @@ impl<Env: Environment> ChainClient<Env> {
             )
         );
         let info = self.chain_info_with_committees().await?;
-        let timestamp = self.next_timestamp(&incoming_bundles, info.timestamp);
-        let transactions = incoming_bundles
-            .into_iter()
-            .map(Transaction::ReceiveMessages)
-            .chain(operations.into_iter().map(Transaction::ExecuteOperation))
-            .collect::<Vec<_>>();
+        let timestamp = self.next_timestamp(&transactions, info.timestamp);
         let proposed_block = ProposedBlock {
             epoch: info.epoch,
             chain_id: self.chain_id,
@@ -2805,22 +2776,19 @@ impl<Env: Environment> ChainClient<Env> {
         self.update_state(|state| {
             state.set_pending_proposal(proposed_block.clone(), blobs.clone())
         });
-        Ok(ConfirmedBlock::new(block))
+        Ok(block)
     }
 
     /// Returns a suitable timestamp for the next block.
     ///
     /// This will usually be the current time according to the local clock, but may be slightly
     /// ahead to make sure it's not earlier than the incoming messages or the previous block.
-    #[instrument(level = "trace", skip(incoming_bundles))]
-    fn next_timestamp(
-        &self,
-        incoming_bundles: &[IncomingBundle],
-        block_time: Timestamp,
-    ) -> Timestamp {
+    #[instrument(level = "trace", skip(transactions))]
+    fn next_timestamp(&self, transactions: &[Transaction], block_time: Timestamp) -> Timestamp {
         let local_time = self.storage_client().clock().current_time();
-        incoming_bundles
+        transactions
             .iter()
+            .filter_map(Transaction::incoming_bundle)
             .map(|msg| msg.bundle.timestamp)
             .max()
             .map_or(local_time, |timestamp| timestamp.max(local_time))
@@ -2948,11 +2916,11 @@ impl<Env: Environment> ChainClient<Env> {
             return Ok((chain_balance, Some(owner_balance)));
         }
         let info = self.chain_info().await?;
-        let timestamp = self.next_timestamp(&incoming_bundles, info.timestamp);
         let transactions = incoming_bundles
             .into_iter()
             .map(Transaction::ReceiveMessages)
             .collect::<Vec<_>>();
+        let timestamp = self.next_timestamp(&transactions, info.timestamp);
         let block = ProposedBlock {
             epoch: info.epoch,
             chain_id: self.chain_id,
@@ -3713,20 +3681,20 @@ impl<Env: Environment> ChainClient<Env> {
         #[cfg(with_metrics)]
         let _latency = metrics::PROCESS_INBOX_WITHOUT_PREPARE_LATENCY.measure_latency();
 
-        let mut epoch_change_ops = self.collect_epoch_changes().await?.into_iter();
-
         let mut certificates = Vec::new();
         loop {
-            let incoming_bundles = self.pending_message_bundles().await?;
-            let stream_updates = self.collect_stream_updates().await?;
-            let block_operations = stream_updates
-                .into_iter()
-                .chain(epoch_change_ops.next())
-                .collect::<Vec<_>>();
-            if incoming_bundles.is_empty() && block_operations.is_empty() {
+            // We provide no operations - this means that the only operations executed
+            // will be epoch changes, receiving messages and processing event stream
+            // updates, if any are pending.
+            let transactions = self.prepend_epochs_messages_and_events(vec![]).await?;
+            // Nothing in the inbox and no stream updates to be processed.
+            if transactions.is_empty() {
                 return Ok((certificates, None));
             }
-            match self.execute_block(block_operations, vec![]).await {
+            match self
+                .execute_prepared_transactions(transactions, vec![])
+                .await
+            {
                 Ok(ExecuteBlockOutcome::Executed(certificate))
                 | Ok(ExecuteBlockOutcome::Conflict(certificate)) => certificates.push(certificate),
                 Ok(ExecuteBlockOutcome::WaitForTimeout(timeout)) => {