Clients update validators about missing committees. (#4493)

## Motivation

If a validator doesn't have the latest committee and receives a block
signed by that committee, it will fail with `EventsNotFound`.

Also, the validator updater is sending the wrong blocks in some cases
for sparse event publisher chains.

## Proposal

Handle that in the client and update the validator about the admin
chain. Use the correct block heights.

Almost all the credit for this goes to @bart-linera!

## Test Plan

A test was added that reproduced the issue.

## Release Plan

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

## Links

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

---------

Co-authored-by: Bartłomiej Kamiński <[email protected]>
Co-authored-by: deuszx <[email protected]>

Author: 3 people

linera-base/src/data_types.rs

@@ -1044,7 +1044,12 @@ pub enum OracleResponse {
     /// The block's validation round.
     Round(Option<u32>),
     /// An event was read.
-    Event(EventId, Vec<u8>),
+    Event(
+        EventId,
+        #[debug(with = "hex_debug")]
+        #[serde(with = "serde_bytes")]
+        Vec<u8>,
+    ),
     /// An event exists.
     EventExists(EventId),
 }

linera-chain/src/chain.rs

@@ -537,6 +537,16 @@ where
         }
     }
 
+    /// Returns the height of the highest block we have, plus one. Includes preprocessed blocks.
+    ///
+    /// The "+ 1" is so that it can be used in the same places as `next_block_height`.
+    pub async fn next_height_to_preprocess(&self) -> Result<BlockHeight, ChainError> {
+        if let Some(height) = self.preprocessed_blocks.indices().await?.last() {
+            return Ok(height.saturating_add(BlockHeight(1)));
+        }
+        Ok(self.tip_state.get().next_block_height)
+    }
+
     pub async fn last_anticipated_block_height(
         &self,
         origin: &ChainId,

linera-core/src/client/mod.rs

@@ -590,7 +590,6 @@ impl<Env: Environment> Client<Env> {
         height: BlockHeight,
         delivery: CrossChainMessageDelivery,
     ) -> Result<(), ChainClientError> {
-        let local_node = self.local_node.clone();
         let nodes = self.make_nodes(committee)?;
         communicate_with_quorum(
             &nodes,
@@ -599,7 +598,8 @@ impl<Env: Environment> Client<Env> {
             |remote_node| {
                 let mut updater = ValidatorUpdater {
                     remote_node,
-                    local_node: local_node.clone(),
+                    local_node: self.local_node.clone(),
+                    admin_id: self.admin_id,
                 };
                 Box::pin(async move {
                     updater
@@ -625,7 +625,6 @@ impl<Env: Environment> Client<Env> {
         action: CommunicateAction,
         value: T,
     ) -> Result<GenericCertificate<T>, ChainClientError> {
-        let local_node = self.local_node.clone();
         let nodes = self.make_nodes(committee)?;
         let ((votes_hash, votes_round), votes) = communicate_with_quorum(
             &nodes,
@@ -634,7 +633,8 @@ impl<Env: Environment> Client<Env> {
             |remote_node| {
                 let mut updater = ValidatorUpdater {
                     remote_node,
-                    local_node: local_node.clone(),
+                    local_node: self.local_node.clone(),
+                    admin_id: self.admin_id,
                 };
                 let action = action.clone();
                 Box::pin(async move { updater.send_chain_update(action).await })

linera-core/src/local_node.rs

@@ -155,7 +155,7 @@ where
         Ok(storage.read_blobs(blob_ids).await?.into_iter().collect())
     }
 
-    /// Reads blob states from storage
+    /// Reads blob states from storage.
     pub async fn read_blob_states_from_storage(
         &self,
         blob_ids: &[BlobId],

linera-core/src/unit_tests/client_tests.rs

@@ -2675,3 +2675,109 @@ where
     assert_eq!(client.local_balance().await.unwrap(), expected_balance);
     Ok(())
 }
+
+#[test_case(MemoryStorageBuilder::default(); "memory")]
+#[cfg_attr(feature = "storage-service", test_case(ServiceStorageBuilder::new().await; "storage_service"))]
+#[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_validator_outdated_admin_chain<B>(storage_builder: B) -> anyhow::Result<()>
+where
+    B: StorageBuilder,
+{
+    let mut signer = InMemorySigner::new(None);
+    let new_public_key = signer.generate_new();
+    let mut builder = TestBuilder::new(storage_builder, 4, 0, signer).await?;
+
+    let admin_client = builder.add_root_chain(0, Amount::from_tokens(1000)).await?;
+    let client1 = builder.add_root_chain(1, Amount::from_tokens(1000)).await?;
+
+    // Take one validator down - they will miss committee changes.
+    builder.set_fault_type([3], FaultType::Offline);
+
+    // Start by creating a block in epoch 0.
+    let certificate0 = client1
+        .transfer(
+            AccountOwner::CHAIN,
+            Amount::ONE,
+            Account::chain(admin_client.chain_id()),
+        )
+        .await
+        .unwrap_ok_committed();
+
+    assert_eq!(certificate0.block().header.epoch, Epoch::from(0));
+
+    // Advance the epoch.
+    admin_client
+        .stage_new_committee(builder.initial_committee.clone())
+        .await
+        .unwrap();
+
+    // Process the inbox to migrate the client's chain.
+    client1.synchronize_from_validators().await.unwrap();
+    client1.process_inbox().await.unwrap();
+
+    // Open a chain
+    let (new_chain_desc, certificate1) = client1
+        .open_chain(
+            ChainOwnership::single(new_public_key.into()),
+            ApplicationPermissions::default(),
+            Amount::from_tokens(10),
+        )
+        .await
+        .unwrap_ok_committed();
+
+    // Check that the epoch has been migrated.
+    assert_eq!(certificate1.block().header.epoch, Epoch::from(1));
+
+    // Make a client to try the new chain.
+    let mut client2 = builder
+        .make_client(new_chain_desc.id(), None, BlockHeight::ZERO)
+        .await?;
+    client2.set_preferred_owner(new_public_key.into());
+    client2.synchronize_from_validators().await.unwrap();
+    client2
+        .receive_certificate_and_update_validators(certificate1)
+        .await
+        .unwrap();
+
+    // Let's deactivate another validator and reactivate the one that was offline.
+    // Now the client will have to update validator 3 on the admin chain in order for the
+    // next blocks to be correctly processed.
+    builder.set_fault_type([2], FaultType::Offline);
+    builder.set_fault_type([3], FaultType::Honest);
+
+    let admin_tip = builder
+        .node(3)
+        .chain_info_with_manager_values(admin_client.chain_id())
+        .await
+        .unwrap()
+        .next_block_height;
+    // At this point, validator 3 should have zero blocks on the admin chain.
+    assert_eq!(admin_tip, 0.into());
+
+    // Update the validators on the chain.
+    // If it works, it means the validator has been correctly updated.
+    client2.update_validators(None).await.unwrap();
+
+    client2
+        .transfer(
+            AccountOwner::CHAIN,
+            Amount::from_tokens(3),
+            Account::chain(admin_client.chain_id()),
+        )
+        .await
+        .unwrap_ok_committed();
+
+    let admin_tip = builder
+        .node(3)
+        .chain_info_with_manager_values(admin_client.chain_id())
+        .await
+        .unwrap()
+        .next_block_height;
+    // Validator 3 should be up to date on the admin chain.
+    assert_eq!(admin_tip, 2.into());
+
+    Ok(())
+}