Extend reconfig test; sync on wallet init; fix blob fetching. (#4437)

## Motivation

We want to make sure the network works fine even after many
reconfigurations, and even if none of the genesis validators are left.

## Proposal

At the end of the reconfiguration end-to-end test, initialize a new
client, request a chain and publish a blob. This test actually failed!
The following two changes make it pass:

* In `wallet init`, synchronize the admin chain, trusting the validators
provided by the faucet (part of
#4434).
* Fix a bug in `process_certificates`: Since we are handling a
_certificate_, not a block proposal, we don't need additional proof for
the missing blobs: The certificate itself is proof for them. So we
should only download the blobs themselves, and not try to get another
certificate for them.

## Test Plan

The reconfiguration test was extended.

## Release Plan

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

## Links

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

Author: afck

linera-core/src/client/mod.rs

@@ -376,18 +376,11 @@ impl<Env: Environment> Client<Env> {
             let mut result = self.handle_certificate(certificate.clone()).await;
 
             if let Err(LocalNodeError::BlobsNotFound(blob_ids)) = &result {
-                future::try_join_all(blob_ids.iter().map(|blob_id| async move {
-                    let blob_certificate =
-                        remote_node.download_certificate_for_blob(*blob_id).await?;
-                    self.receive_sender_certificate(
-                        blob_certificate,
-                        ReceiveCertificateMode::NeedsCheck,
-                        None,
-                    )
-                    .await?;
-                    Result::<(), ChainClientError>::Ok(())
+                let blobs = future::join_all(blob_ids.iter().map(|blob_id| async move {
+                    remote_node.try_download_blob(*blob_id).await.unwrap()
                 }))
-                .await?;
+                .await;
+                self.local_node.store_blobs(&blobs).await?;
                 result = self.handle_certificate(certificate.clone()).await;
             }
 
@@ -930,11 +923,23 @@ impl<Env: Environment> Client<Env> {
     async fn synchronize_chain_state(
         &self,
         chain_id: ChainId,
+    ) -> Result<Box<ChainInfo>, ChainClientError> {
+        let (_, committee) = self.admin_committee().await?;
+        self.synchronize_chain_state_from_committee(chain_id, committee)
+            .await
+    }
+
+    /// Downloads and processes any certificates we are missing for the given chain, from the given
+    /// committee.
+    #[instrument(level = "trace", skip_all)]
+    pub async fn synchronize_chain_state_from_committee(
+        &self,
+        chain_id: ChainId,
+        committee: Committee,
     ) -> Result<Box<ChainInfo>, ChainClientError> {
         #[cfg(with_metrics)]
         let _latency = metrics::SYNCHRONIZE_CHAIN_STATE_LATENCY.measure_latency();
 
-        let (_, committee) = self.admin_committee().await?;
         let validators = self.make_nodes(&committee)?;
         Box::pin(self.fetch_chain_info(chain_id, &validators)).await?;
         communicate_with_quorum(
@@ -2297,6 +2302,18 @@ impl<Env: Environment> ChainClient<Env> {
         self.client.synchronize_chain_state(chain_id).await
     }
 
+    /// Downloads and processes any certificates we are missing for this chain, from the given
+    /// committee.
+    #[instrument(level = "trace", skip_all)]
+    pub async fn synchronize_chain_state_from_committee(
+        &self,
+        committee: Committee,
+    ) -> Result<Box<ChainInfo>, ChainClientError> {
+        self.client
+            .synchronize_chain_state_from_committee(self.chain_id, committee)
+            .await
+    }
+
     /// Executes a list of operations.
     #[instrument(level = "trace", skip(operations, blobs))]
     pub async fn execute_operations(

linera-core/src/remote_node.rs

@@ -210,7 +210,7 @@ impl<N: ValidatorNode> RemoteNode<N> {
     }
 
     #[instrument(level = "trace")]
-    async fn try_download_blob(&self, blob_id: BlobId) -> Option<Blob> {
+    pub async fn try_download_blob(&self, blob_id: BlobId) -> Option<Blob> {
         match self.node.download_blob(blob_id).await {
             Ok(blob) => {
                 let blob = Blob::new(blob);

linera-service/src/cli/main.rs

@@ -19,7 +19,7 @@ use chrono::Utc;
 use colored::Colorize;
 use futures::{lock::Mutex, FutureExt as _, StreamExt};
 use linera_base::{
-    crypto::{InMemorySigner, Signer},
+    crypto::{AccountPublicKey, InMemorySigner, Signer},
     data_types::{ApplicationPermissions, Timestamp},
     identifiers::{AccountOwner, ChainId},
     listen_for_shutdown_signals,
@@ -1610,6 +1610,43 @@ impl Runnable for Job {
                 );
             }
 
+            Wallet(WalletCommand::Init { faucet, .. }) => {
+                let Some(faucet_url) = faucet else {
+                    return Ok(());
+                };
+                let Some(network_description) = storage.read_network_description().await? else {
+                    anyhow::bail!("Missing network description");
+                };
+                let context = ClientContext::new(
+                    storage,
+                    options.context_options.clone(),
+                    wallet,
+                    signer.into_value(),
+                );
+                let faucet = cli_wrappers::Faucet::new(faucet_url);
+                let validators = faucet.current_validators().await?;
+                let chain_client = context.make_chain_client(network_description.admin_chain_id);
+                // TODO(#4434): This is a quick workaround with an equal-weight committee. Instead,
+                // the faucet should provide the full committee including weights.
+                let committee = Committee::new(
+                    validators
+                        .into_iter()
+                        .map(|(pub_key, network_address)| {
+                            let state = ValidatorState {
+                                network_address,
+                                votes: 100,
+                                account_public_key: AccountPublicKey::from_slice(&[0; 33]).unwrap(),
+                            };
+                            (pub_key, state)
+                        })
+                        .collect(),
+                    Default::default(), // unused
+                );
+                chain_client
+                    .synchronize_chain_state_from_committee(committee)
+                    .await?;
+            }
+
             Wallet(WalletCommand::FollowChain {
                 chain_id,
                 sync: true,
@@ -2431,6 +2468,7 @@ Make sure to use a Linera client compatible with this network.
                 keystore.persist().await?;
                 options.create_wallet(genesis_config)?.persist().await?;
                 options.initialize_storage().boxed().await?;
+                options.run_with_storage(Job(options.clone())).await??;
                 info!(
                     "Wallet initialized in {} ms",
                     start_time.elapsed().as_millis()

linera-service/tests/local_net_tests.rs

@@ -237,6 +237,19 @@ async fn test_end_to_end_reconfiguration(config: LocalNetConfig) -> Result<()> {
         );
     }
 
+    if matches!(network, Network::Grpc) {
+        let client = net.make_client().await;
+        client.wallet_init(Some(&faucet)).await?;
+        let (chain_id, _owner) = client.request_chain(&faucet, true).await?;
+        let port = get_node_port().await;
+        let service = client
+            .run_node_service(port, ProcessInbox::Automatic)
+            .await?;
+        service
+            .publish_data_blob(&chain_id, b"blob bytes".to_vec())
+            .await?;
+    }
+
     net.ensure_is_running().await?;
     net.terminate().await?;