Preparations for epoch change events. (#3431)

## Motivation

To make epoch changes scalable, we want to use events instead of
messages (#365).

This will be done in an upcoming PR. To make that easier to review, I
separated some of the changes I need to make as a preparation.

The follow-up PR will add `ProcessNewEpoch` and `ProcessRemovedEpoch`
system operations and an `Oracle::Response::Event` variant. Clients will
be _notified_ about new epochs because they will always
follow/synchronize the admin chain and thus store the epoch change
events locally. They will then run `process_inbox` which, in addition to
incoming messages, will look for epoch change events, and include the
new operations in the proposal accordingly. Processing the operations
will check for the event, add the oracle response, and update the
chain's epoch and committees.

## Proposal

* Add an `Event` oracle response variant, to record read event values in
the block.
* Don't fail in `make_chain_client` if the admin chain is not in the
wallet. Clients will always need to sync/follow the admin chain to learn
about epoch change events.
* Return early from `try_synchronize_chain_state_from` if the remote
node has no new blocks for us.
* Always initialize the root chain in the `linera-core` tests, even if
it is not mentioned in the test case.
* Add events methods to the execution runtime context: We will need
these in `system.rs` to load the epoch change events.
* Add more assertions to `test_end_to_end_reconfiguration`. This was
useful for debugging.
* Other minor cleanups.

## Test Plan

No new behavior was added. CI will catch regressions.

## Release Plan

- Nothing to do / These changes follow the usual release cycle.

## Links

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

Author: afck

linera-client/src/client_context.rs

@@ -238,23 +238,22 @@ where
     }
 
     fn make_chain_client(&self, chain_id: ChainId) -> Result<ChainClient<NodeProvider, S>, Error> {
-        let chain = self
-            .wallet
-            .get(chain_id)
-            .ok_or_else(|| error::Inner::NonexistentChain(chain_id))?;
-        let known_key_pairs = chain
-            .key_pair
-            .as_ref()
-            .map(|kp| kp.copy())
-            .into_iter()
-            .collect();
+        // We only create clients for chains we have in the wallet, or for the admin chain.
+        let chain = match self.wallet.get(chain_id) {
+            Some(chain) => chain.clone(),
+            None if chain_id == self.wallet.genesis_admin_chain() => {
+                UserChain::make_other(self.wallet.genesis_admin_chain(), Timestamp::from(0))
+            }
+            None => return Err(error::Inner::NonexistentChain(chain_id).into()),
+        };
+        let known_key_pairs = chain.key_pair.into_iter().collect();
         Ok(self.make_chain_client_internal(
             chain_id,
             known_key_pairs,
             chain.block_hash,
             chain.timestamp,
             chain.next_block_height,
-            chain.pending_proposal.clone(),
+            chain.pending_proposal,
         ))
     }
 

linera-client/src/wallet.rs

@@ -213,6 +213,19 @@ pub struct UserChain {
     pub pending_proposal: Option<PendingProposal>,
 }
 
+impl Clone for UserChain {
+    fn clone(&self) -> Self {
+        Self {
+            chain_id: self.chain_id,
+            key_pair: self.key_pair.as_ref().map(AccountSecretKey::copy),
+            block_hash: self.block_hash,
+            timestamp: self.timestamp,
+            next_block_height: self.next_block_height,
+            pending_proposal: self.pending_proposal.clone(),
+        }
+    }
+}
+
 impl UserChain {
     /// Create a user chain that we own.
     pub fn make_initial<R: CryptoRng>(

linera-core/src/chain_worker/state/attempted_changes.rs

@@ -353,7 +353,7 @@ where
                 .events
                 .iter()
                 .flatten()
-                .map(|event| (event.id(chain_id), &event.value[..]));
+                .map(|event| (event.id(chain_id), event.value.clone()));
             self.state.storage.write_events(events).await?;
         }
 

linera-core/src/client/mod.rs

@@ -1011,8 +1011,7 @@ where
             // For chains with any owner other than ourselves, we could be missing recent
             // certificates created by other owners. Further synchronize blocks from the network.
             // This is a best-effort that depends on network conditions.
-            let nodes = self.validator_nodes().await?;
-            info = self.synchronize_chain_state(&nodes, self.chain_id).await?;
+            info = self.synchronize_chain_state(self.chain_id).await?;
         }
 
         let result = self
@@ -1700,17 +1699,20 @@ where
 
     /// Downloads and processes any certificates we are missing for the given chain.
     #[instrument(level = "trace", skip_all)]
-    pub async fn synchronize_chain_state(
+    async fn synchronize_chain_state(
         &self,
-        validators: &[RemoteNode<P::Node>],
         chain_id: ChainId,
     ) -> Result<Box<ChainInfo>, ChainClientError> {
         #[cfg(with_metrics)]
         let _latency = metrics::SYNCHRONIZE_CHAIN_STATE_LATENCY.measure_latency();
 
-        let committee = self.local_committee().await?;
+        let (epoch, mut committees) = self.epoch_and_committees(chain_id).await?;
+        let committee = committees
+            .remove(&epoch.ok_or(LocalNodeError::InvalidChainInfoResponse)?)
+            .ok_or(LocalNodeError::InvalidChainInfoResponse)?;
+        let validators = self.make_nodes(&committee)?;
         communicate_with_quorum(
-            validators,
+            &validators,
             &committee,
             |_: &()| (),
             |remote_node| {
@@ -1749,6 +1751,9 @@ where
             .with_sent_certificate_hashes_in_range(range)
             .with_manager_values();
         let info = remote_node.handle_chain_info_query(query).await?;
+        if info.next_block_height < local_info.next_block_height {
+            return Ok(());
+        }
 
         let certificates: Vec<ConfirmedBlockCertificate> = remote_node
             .download_certificates(info.requested_sent_certificate_hashes)
@@ -2442,9 +2447,7 @@ where
     pub async fn synchronize_from_validators(&self) -> Result<Box<ChainInfo>, ChainClientError> {
         if self.chain_id != self.admin_id {
             // Synchronize the state of the admin chain from the network.
-            let local_committee = self.local_committee().await?;
-            let nodes = self.make_nodes(&local_committee)?;
-            self.synchronize_chain_state(&nodes, self.admin_id).await?;
+            self.synchronize_chain_state(self.admin_id).await?;
         }
         let info = self.prepare_chain().await?;
         self.find_received_certificates().await?;

linera-core/src/unit_tests/test_utils.rs

@@ -797,11 +797,18 @@ where
     }
 
     /// Creates the root chain with the given `index`, and returns a client for it.
+    ///
+    /// Root chain 0 is the admin chain and needs to be initialized first, otherwise its balance
+    /// is automatically set to zero.
     pub async fn add_root_chain(
         &mut self,
         index: u32,
         balance: Amount,
     ) -> Result<ChainClient<NodeProvider<B::Storage>, B::Storage>, anyhow::Error> {
+        // Make sure the admin chain is initialized.
+        if self.genesis_storage_builder.accounts.is_empty() && index != 0 {
+            Box::pin(self.add_root_chain(0, Amount::ZERO)).await?;
+        }
         let description = ChainDescription::Root(index);
         let key_pair = AccountSecretKey::generate();
         let public_key = key_pair.public();

linera-execution/src/lib.rs

@@ -44,7 +44,8 @@ use linera_base::{
     doc_scalar, hex_debug, http,
     identifiers::{
         Account, AccountOwner, ApplicationId, BlobId, BytecodeId, ChainId, ChannelName,
-        Destination, GenericApplicationId, MessageId, Owner, StreamName, UserApplicationId,
+        Destination, EventId, GenericApplicationId, MessageId, Owner, StreamName,
+        UserApplicationId,
     },
     ownership::ChainOwnership,
     task,
@@ -392,13 +393,21 @@ pub trait ExecutionRuntimeContext {
 
     async fn get_blob(&self, blob_id: BlobId) -> Result<Blob, ViewError>;
 
+    async fn get_event(&self, event_id: EventId) -> Result<Vec<u8>, ViewError>;
+
     async fn contains_blob(&self, blob_id: BlobId) -> Result<bool, ViewError>;
 
     #[cfg(with_testing)]
     async fn add_blobs(
         &self,
         blobs: impl IntoIterator<Item = Blob> + Send,
     ) -> Result<(), ViewError>;
+
+    #[cfg(with_testing)]
+    async fn add_events(
+        &self,
+        events: impl IntoIterator<Item = (EventId, Vec<u8>)> + Send,
+    ) -> Result<(), ViewError>;
 }
 
 #[derive(Clone, Copy, Debug)]
@@ -1052,6 +1061,7 @@ pub struct TestExecutionRuntimeContext {
     user_contracts: Arc<DashMap<UserApplicationId, UserContractCode>>,
     user_services: Arc<DashMap<UserApplicationId, UserServiceCode>>,
     blobs: Arc<DashMap<BlobId, Blob>>,
+    events: Arc<DashMap<EventId, Vec<u8>>>,
 }
 
 #[cfg(with_testing)]
@@ -1063,6 +1073,7 @@ impl TestExecutionRuntimeContext {
             user_contracts: Arc::default(),
             user_services: Arc::default(),
             blobs: Arc::default(),
+            events: Arc::default(),
         }
     }
 }
@@ -1123,6 +1134,14 @@ impl ExecutionRuntimeContext for TestExecutionRuntimeContext {
             .clone())
     }
 
+    async fn get_event(&self, event_id: EventId) -> Result<Vec<u8>, ViewError> {
+        Ok(self
+            .events
+            .get(&event_id)
+            .ok_or_else(|| ViewError::EventsNotFound(vec![event_id]))?
+            .clone())
+    }
+
     async fn contains_blob(&self, blob_id: BlobId) -> Result<bool, ViewError> {
         Ok(self.blobs.contains_key(&blob_id))
     }
@@ -1138,6 +1157,18 @@ impl ExecutionRuntimeContext for TestExecutionRuntimeContext {
 
         Ok(())
     }
+
+    #[cfg(with_testing)]
+    async fn add_events(
+        &self,
+        events: impl IntoIterator<Item = (EventId, Vec<u8>)> + Send,
+    ) -> Result<(), ViewError> {
+        for (event_id, bytes) in events {
+            self.events.insert(event_id, bytes);
+        }
+
+        Ok(())
+    }
 }
 
 impl From<SystemOperation> for Operation {

linera-sdk/Cargo.toml

@@ -48,6 +48,9 @@ test = [
 [dependencies]
 async-graphql.workspace = true
 async-trait = { workspace = true, optional = true }
+
+# TODO(#3421): Remove the pinned version once the `linera-*` crates move to Rust Edition 2024
+base64ct = "=1.6.0"
 bcs.workspace = true
 futures.workspace = true
 linera-base.workspace = true
@@ -60,9 +63,6 @@ serde_json.workspace = true
 thiserror.workspace = true
 wit-bindgen.workspace = true
 
-# TODO(#3421): Remove the pinned version once the `linera-*` crates move to Rust Edition 2024
-base64ct = "=1.6.0"
-
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
 async-graphql.workspace = true
 anyhow.workspace = true

linera-service/src/cli_wrappers/wallet.rs

@@ -3,7 +3,7 @@
 
 use std::{
     borrow::Cow,
-    collections::HashMap,
+    collections::{BTreeMap, HashMap},
     env,
     marker::PhantomData,
     mem,
@@ -26,7 +26,11 @@ use linera_base::{
 };
 use linera_client::wallet::Wallet;
 use linera_core::worker::Notification;
-use linera_execution::{system::SystemChannel, ResourceControlPolicy};
+use linera_execution::{
+    committee::{Committee, Epoch},
+    system::SystemChannel,
+    ResourceControlPolicy,
+};
 use linera_faucet::ClaimOutcome;
 use linera_faucet_client::Faucet;
 use serde::{de::DeserializeOwned, ser::Serialize};
@@ -1239,6 +1243,17 @@ impl NodeService {
         Ok(bytecode_id.with_abi())
     }
 
+    pub async fn query_committees(&self, chain_id: &ChainId) -> Result<BTreeMap<Epoch, Committee>> {
+        let query = format!(
+            "query {{ chain(chainId:\"{chain_id}\") {{
+                executionState {{ system {{ committees }} }}
+            }} }}"
+        );
+        let mut response = self.query_node(query).await?;
+        let committees = response["chain"]["executionState"]["system"]["committees"].take();
+        Ok(serde_json::from_value(committees)?)
+    }
+
     pub async fn query_node(&self, query: impl AsRef<str>) -> Result<Value> {
         let n_try = 5;
         let query = query.as_ref();

linera-service/src/proxy/grpc.rs

@@ -352,6 +352,7 @@ where
             }
             ViewError::NotFound(_)
             | ViewError::BlobsNotFound(_)
+            | ViewError::EventsNotFound(_)
             | ViewError::CannotAcquireCollectionEntry
             | ViewError::MissingEntries => Status::not_found(err.to_string()),
         };

linera-service/tests/local_net_tests.rs

@@ -20,6 +20,7 @@ use linera_base::{
     identifiers::{Account, AccountOwner, ChainId},
 };
 use linera_core::{data_types::ChainInfoQuery, node::ValidatorNode};
+use linera_execution::committee::Epoch;
 use linera_faucet::ClaimOutcome;
 use linera_sdk::base::AccountSecretKey;
 use linera_service::{
@@ -148,6 +149,9 @@ async fn test_end_to_end_reconfiguration(config: LocalNetConfig) -> Result<()> {
     client.query_validators(Some(chain_1)).await?;
     if let Some(service) = &node_service_2 {
         service.process_inbox(&chain_2).await?;
+        let committees = service.query_committees(&chain_2).await?;
+        let epochs = committees.into_keys().collect::<Vec<_>>();
+        assert_eq!(&epochs, &[Epoch(3)]);
     } else {
         client_2.process_inbox(chain_2).await?;
     }
@@ -159,6 +163,9 @@ async fn test_end_to_end_reconfiguration(config: LocalNetConfig) -> Result<()> {
         client.finalize_committee().await?;
         if let Some(service) = &node_service_2 {
             service.process_inbox(&chain_2).await?;
+            let committees = service.query_committees(&chain_2).await?;
+            let epochs = committees.into_keys().collect::<Vec<_>>();
+            assert_eq!(&epochs, &[Epoch(4 + i as u32)]);
         } else {
             client_2.process_inbox(chain_2).await?;
         }
@@ -174,30 +181,23 @@ async fn test_end_to_end_reconfiguration(config: LocalNetConfig) -> Result<()> {
         )
         .await?;
 
-    if let Some(mut node_service_2) = node_service_2 {
-        node_service_2.process_inbox(&chain_2).await?;
+    if let Some(mut service) = node_service_2 {
+        service.process_inbox(&chain_2).await?;
         let query = format!(
             "query {{ chain(chainId:\"{chain_2}\") {{
                 executionState {{ system {{ balances {{
                     entry(key:\"{recipient}\") {{ value }}
                 }} }} }}
             }} }}"
         );
-        let response = node_service_2.query_node(query.clone()).await?;
+        let response = service.query_node(query).await?;
         let balances = &response["chain"]["executionState"]["system"]["balances"];
         assert_eq!(balances["entry"]["value"].as_str(), Some("5."));
+        let committees = service.query_committees(&chain_2).await?;
+        let epochs = committees.into_keys().collect::<Vec<_>>();
+        assert_eq!(&epochs, &[Epoch(7)]);
 
-        let query = format!(
-            "query {{ chain(chainId:\"{chain_2}\") {{
-                executionState {{ system {{ committees }} }}
-            }} }}"
-        );
-        let response = node_service_2.query_node(query.clone()).await?;
-        let committees = &response["chain"]["executionState"]["system"]["committees"];
-        let epochs = committees.as_object().unwrap().keys().collect::<Vec<_>>();
-        assert_eq!(&epochs, &["7"]);
-
-        node_service_2.ensure_is_running()?;
+        service.ensure_is_running()?;
     } else {
         client_2.sync(chain_2).await?;
         client_2.process_inbox(chain_2).await?;

linera-storage/src/db_storage.rs

@@ -216,7 +216,7 @@ trait BatchExt {
     fn add_certificate(&mut self, certificate: &ConfirmedBlockCertificate)
         -> Result<(), ViewError>;
 
-    fn add_event(&mut self, event_id: EventId, value: &[u8]) -> Result<(), ViewError>;
+    fn add_event(&mut self, event_id: EventId, value: Vec<u8>) -> Result<(), ViewError>;
 }
 
 impl BatchExt for Batch {
@@ -248,11 +248,11 @@ impl BatchExt for Batch {
         Ok(())
     }
 
-    fn add_event(&mut self, event_id: EventId, value: &[u8]) -> Result<(), ViewError> {
+    fn add_event(&mut self, event_id: EventId, value: Vec<u8>) -> Result<(), ViewError> {
         #[cfg(with_metrics)]
         WRITE_EVENT_COUNTER.with_label_values(&[]).inc();
         let event_key = bcs::to_bytes(&BaseKey::Event(event_id))?;
-        self.put_key_value_bytes(event_key.to_vec(), value.to_vec());
+        self.put_key_value_bytes(event_key.to_vec(), value);
         Ok(())
     }
 }
@@ -809,15 +809,15 @@ where
 
     async fn read_event(&self, event_id: EventId) -> Result<Vec<u8>, ViewError> {
         let event_key = bcs::to_bytes(&BaseKey::Event(event_id.clone()))?;
-        let maybe_value = self.store.read_value::<Vec<u8>>(&event_key).await?;
+        let maybe_value = self.store.read_value_bytes(&event_key).await?;
         #[cfg(with_metrics)]
         READ_EVENT_COUNTER.with_label_values(&[]).inc();
-        maybe_value.ok_or_else(|| ViewError::not_found("value for event ID", event_id))
+        maybe_value.ok_or_else(|| ViewError::EventsNotFound(vec![event_id]))
     }
 
     async fn write_events(
         &self,
-        events: impl IntoIterator<Item = (EventId, &[u8])> + Send,
+        events: impl IntoIterator<Item = (EventId, Vec<u8>)> + Send,
     ) -> Result<(), ViewError> {
         let mut batch = Batch::new();
         for (event_id, value) in events {