Add linera sync-validator CLI command (#3156)

## Motivation

<!--
Briefly describe the goal(s) of this PR.
-->
Some testnet validators started lagging, and that made clients stop
sending certificates to them, which made them lag even more.

## Proposal

<!--
Summarize the proposed changes and how they address the goal(s) stated
above.
-->
As a quick-fix, a `linera sync-validator` CLI command was added that
pushes local certificates that are missing from a specified validator.

## Test Plan

<!--
Explain how you made sure that the changes are correct and that they
perform as intended.

Please describe testing protocols (CI, manual tests, benchmarks, etc) in
a way that others
can reproduce the results.
-->
A simple end-to-end test was written to ensure that it behaves as
expected.

## Release Plan

<!--
If this PR targets the `main` branch, **keep the applicable lines** to
indicate if you
recommend the changes to be picked in release branches, SDKs, and
hotfixes.

This generally concerns only bug fixes.

Note that altering the public protocol (e.g. transaction format, WASM
syscalls) or storage
formats requires a new deployment.
-->
- Nothing to do / These changes follow the usual release cycle, because
it adds a new feature without breaking any backward compatibility.

## Links

<!--
Optional section for related PRs, related issues, and other references.

If needed, please create issues to track future improvements and link
them here.
-->
- [reviewer
checklist](https://github.com/linera-io/linera-protocol/blob/main/CONTRIBUTING.md#reviewer-checklist)

CLI.md

@@ -18,6 +18,7 @@ This document contains the help content for the `linera` command-line program.
 * [`linera process-inbox`↴](#linera-process-inbox)
 * [`linera query-validator`↴](#linera-query-validator)
 * [`linera query-validators`↴](#linera-query-validators)
+* [`linera sync-validator`↴](#linera-sync-validator)
 * [`linera set-validator`↴](#linera-set-validator)
 * [`linera remove-validator`↴](#linera-remove-validator)
 * [`linera finalize-committee`↴](#linera-finalize-committee)
@@ -77,6 +78,7 @@ A Byzantine-fault tolerant sidechain with low-latency finality and high throughp
 * `process-inbox` — Process all pending incoming messages from the inbox of the given chain by creating as many blocks as needed to execute all (non-failing) messages. Failing messages will be marked as rejected and may bounce to their sender depending on their configuration
 * `query-validator` — Show the version and genesis config hash of a new validator, and print a warning if it is incompatible. Also print some information about the given chain while we are at it
 * `query-validators` — Show the current set of validators for a chain. Also print some information about the given chain while we are at it
+* `sync-validator` — Synchronizes a validator with the local state of chains
 * `set-validator` — Add or modify a validator (admin only)
 * `remove-validator` — Remove a validator (admin only)
 * `finalize-committee` — Deprecates all committees except the last one
@@ -380,6 +382,22 @@ Show the current set of validators for a chain. Also print some information abou
 
 
 
+## `linera sync-validator`
+
+Synchronizes a validator with the local state of chains
+
+**Usage:** `linera sync-validator [OPTIONS] <ADDRESS>`
+
+###### **Arguments:**
+
+* `<ADDRESS>` — The public address of the validator to synchronize
+
+###### **Options:**
+
+* `--chains <CHAINS>` — The chains to synchronize, or the default chain if empty
+
+
+
 ## `linera set-validator`
 
 Add or modify a validator (admin only)

linera-client/src/client_options.rs

@@ -463,6 +463,16 @@ pub enum ClientCommand {
         chain_id: Option<ChainId>,
     },
 
+    /// Synchronizes a validator with the local state of chains.
+    SyncValidator {
+        /// The public address of the validator to synchronize.
+        address: String,
+
+        /// The chains to synchronize, or the default chain if empty.
+        #[arg(long, num_args = 0..)]
+        chains: Vec<ChainId>,
+    },
+
     /// Add or modify a validator (admin only)
     SetValidator {
         /// The public key of the validator.

linera-core/src/client/mod.rs

@@ -3426,6 +3426,53 @@ where
             },
         )
     }
+
+    /// Attempts to update a validator with the local information.
+    #[instrument(level = "trace", skip(remote_node))]
+    pub async fn sync_validator(&self, remote_node: P::Node) -> Result<(), ChainClientError> {
+        let validator_chain_state = remote_node
+            .handle_chain_info_query(ChainInfoQuery::new(self.chain_id))
+            .await?;
+        let local_chain_state = self.client.local_node.chain_info(self.chain_id).await?;
+
+        let Some(missing_certificate_count) = local_chain_state
+            .next_block_height
+            .0
+            .checked_sub(validator_chain_state.info.next_block_height.0)
+            .filter(|count| *count > 0)
+        else {
+            debug!("Validator is up-to-date with local state");
+            return Ok(());
+        };
+
+        let missing_certificates_end = usize::try_from(local_chain_state.next_block_height.0)
+            .expect("`usize` should be at least `u64`");
+        let missing_certificates_start = missing_certificates_end
+            - usize::try_from(missing_certificate_count).expect("`usize` should be at least `u64`");
+
+        let missing_certificate_hashes = self
+            .client
+            .local_node
+            .chain_state_view(self.chain_id)
+            .await?
+            .confirmed_log
+            .read(missing_certificates_start..missing_certificates_end)
+            .await?;
+
+        let certificates = self
+            .client
+            .storage
+            .read_certificates(missing_certificate_hashes)
+            .await?;
+
+        for certificate in certificates {
+            remote_node
+                .handle_confirmed_certificate(certificate, CrossChainMessageDelivery::NonBlocking)
+                .await?;
+        }
+
+        Ok(())
+    }
 }
 
 /// The outcome of trying to commit a list of incoming messages and operations to the chain.

linera-rpc/src/lib.rs

@@ -21,8 +21,9 @@ pub mod simple;
 
 pub mod grpc;
 
+pub use client::Client;
 pub use message::RpcMessage;
-pub use node_provider::NodeOptions;
+pub use node_provider::{NodeOptions, NodeProvider};
 
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 #[cfg_attr(with_testing, derive(Eq, PartialEq))]

linera-service/src/cli_wrappers/local_net.rs

@@ -20,6 +20,7 @@ use linera_base::{
     data_types::Amount,
 };
 use linera_client::storage::{StorageConfig, StorageConfigNamespace};
+use linera_core::node::ValidatorNodeProvider;
 use linera_execution::ResourceControlPolicy;
 #[cfg(all(feature = "storage-service", with_testing))]
 use linera_storage_service::common::storage_service_test_endpoint;
@@ -31,7 +32,7 @@ use tonic::transport::{channel::ClientTlsConfig, Endpoint};
 use tonic_health::pb::{
     health_check_response::ServingStatus, health_client::HealthClient, HealthCheckRequest,
 };
-use tracing::{info, warn};
+use tracing::{error, info, warn};
 
 use crate::{
     cli_wrappers::{
@@ -627,6 +628,37 @@ impl LocalNet {
         self.running_validators.insert(validator, validator_proxy);
         Ok(())
     }
+
+    /// Terminates all the processes of a given validator.
+    pub async fn stop_validator(&mut self, validator_index: usize) -> Result<()> {
+        if let Some(mut validator) = self.running_validators.remove(&validator_index) {
+            if let Err(error) = validator.terminate().await {
+                error!("Failed to stop validator {validator_index}: {error}");
+                return Err(error);
+            }
+        }
+        Ok(())
+    }
+
+    /// Returns a [`linera_rpc::Client`] to interact directly with a `validator`.
+    pub async fn validator_client(&mut self, validator: usize) -> Result<linera_rpc::Client> {
+        let node_provider = linera_rpc::NodeProvider::new(linera_rpc::NodeOptions {
+            send_timeout: Duration::from_secs(1),
+            recv_timeout: Duration::from_secs(1),
+            retry_delay: Duration::ZERO,
+            max_retries: 0,
+        });
+
+        Ok(node_provider.make_node(&self.validator_address(validator))?)
+    }
+
+    /// Returns the address to connect to a validator's proxy.
+    pub fn validator_address(&self, validator: usize) -> String {
+        let port = Self::proxy_port(validator);
+        let schema = self.network.external.schema();
+
+        format!("{schema}:localhost:{port}")
+    }
 }
 
 #[cfg(with_testing)]

linera-service/src/cli_wrappers/mod.rs

@@ -111,4 +111,13 @@ impl Network {
             Network::Tcp | Network::Udp => "127.0.0.1",
         }
     }
+
+    /// Returns the protocol schema to use in the node address tuple.
+    pub fn schema(&self) -> &'static str {
+        match self {
+            Network::Grpc | Network::Grpcs => "grpc",
+            Network::Tcp => "tcp",
+            Network::Udp => "udp",
+        }
+    }
 }

linera-service/src/cli_wrappers/wallet.rs

@@ -521,6 +521,24 @@ impl ClientWrapper {
         Ok(())
     }
 
+    /// Runs `linera sync-validator`.
+    pub async fn sync_validator(
+        &self,
+        chain_ids: impl IntoIterator<Item = &ChainId>,
+        validator_address: impl Into<String>,
+    ) -> Result<()> {
+        let mut command = self.command().await?;
+        command.arg("sync-validator").arg(validator_address.into());
+        let mut chain_ids = chain_ids.into_iter().peekable();
+        if chain_ids.peek().is_some() {
+            command
+                .arg("--chains")
+                .args(chain_ids.map(ChainId::to_string));
+        }
+        command.spawn_and_wait_for_stdout().await?;
+        Ok(())
+    }
+
     /// Runs `linera faucet`.
     pub async fn run_faucet(
         &self,

linera-service/src/linera/main.rs

@@ -517,6 +517,23 @@ impl Runnable for Job {
                 println!("{}/{} OK.", num_ok_validators, committee.validators().len());
             }
 
+            SyncValidator {
+                address,
+                mut chains,
+            } => {
+                if chains.is_empty() {
+                    chains.push(context.default_chain());
+                }
+
+                let validator = context.make_node_provider().make_node(&address)?;
+
+                for chain_id in chains {
+                    let chain = context.make_chain_client(chain_id)?;
+
+                    chain.sync_validator(validator.clone()).await?;
+                }
+            }
+
             command @ (SetValidator { .. }
             | RemoveValidator { .. }
             | ResourceControlPolicy { .. }) => {
@@ -1408,6 +1425,7 @@ fn log_file_name_for(command: &ClientCommand) -> Cow<'static, str> {
         | ClientCommand::ProcessInbox { .. }
         | ClientCommand::QueryValidator { .. }
         | ClientCommand::QueryValidators { .. }
+        | ClientCommand::SyncValidator { .. }
         | ClientCommand::SetValidator { .. }
         | ClientCommand::RemoveValidator { .. }
         | ClientCommand::ResourceControlPolicy { .. }

linera-service/tests/local_net_tests.rs

@@ -15,9 +15,10 @@ use std::{env, path::PathBuf, process::Command, time::Duration};
 use anyhow::Result;
 use common::INTEGRATION_TEST_GUARD;
 use linera_base::{
-    data_types::Amount,
+    data_types::{Amount, BlockHeight},
     identifiers::{Account, AccountOwner, ChainId},
 };
+use linera_core::{data_types::ChainInfoQuery, node::ValidatorNode};
 use linera_service::{
     cli_wrappers::{
         local_net::{
@@ -845,3 +846,70 @@ async fn test_end_to_end_benchmark(mut config: LocalNetConfig) -> Result<()> {
 
     Ok(())
 }
+
+/// Tests if the `sync-validator` command uploads missing certificates to a validator.
+// TODO(#3258): Fix test for simple-net
+// #[cfg_attr(feature = "scylladb", test_case(LocalNetConfig::new_test(Database::ScyllaDb, Network::Udp) ; "scylladb_udp"))]
+#[cfg_attr(feature = "scylladb", test_case(LocalNetConfig::new_test(Database::ScyllaDb, Network::Grpc) ; "scylladb_grpc"))]
+#[cfg_attr(feature = "storage-service", test_case(LocalNetConfig::new_test(Database::Service, Network::Grpc) ; "storage_service_grpc"))]
+// #[cfg_attr(feature = "storage-service", test_case(LocalNetConfig::new_test(Database::Service, Network::Tcp) ; "storage_service_tcp"))]
+#[cfg_attr(feature = "dynamodb", test_case(LocalNetConfig::new_test(Database::DynamoDb, Network::Grpc) ; "aws_grpc"))]
+// #[cfg_attr(feature = "scylladb", test_case(LocalNetConfig::new_test(Database::ScyllaDb, Network::Tcp) ; "scylladb_tcp"))]
+// #[cfg_attr(feature = "dynamodb", test_case(LocalNetConfig::new_test(Database::DynamoDb, Network::Tcp) ; "aws_tcp"))]
+// #[cfg_attr(feature = "dynamodb", test_case(LocalNetConfig::new_test(Database::DynamoDb, Network::Udp) ; "aws_udp"))]
+#[test_log::test(tokio::test)]
+async fn test_sync_validator(config: LocalNetConfig) -> Result<()> {
+    let _guard = INTEGRATION_TEST_GUARD.lock().await;
+    tracing::info!("Starting test {}", test_name!());
+
+    const BLOCKS_TO_CREATE: usize = 5;
+    const LAGGING_VALIDATOR_INDEX: usize = 0;
+
+    let (mut net, client) = config.instantiate().await?;
+
+    // Stop a validator to force it to lag behind the others
+    net.stop_validator(LAGGING_VALIDATOR_INDEX).await?;
+
+    // Create some blocks
+    let sender_chain = client.default_chain().expect("Client has no default chain");
+    let (_, receiver_chain) = client
+        .open_chain(sender_chain, None, Amount::from_tokens(1_000))
+        .await?;
+
+    for amount in 1..=BLOCKS_TO_CREATE {
+        client
+            .transfer(
+                Amount::from_tokens(amount as u128),
+                sender_chain,
+                receiver_chain,
+            )
+            .await?;
+    }
+
+    // Restart the stopped validator
+    net.start_validator(LAGGING_VALIDATOR_INDEX).await?;
+
+    let lagging_validator = net.validator_client(LAGGING_VALIDATOR_INDEX).await?;
+
+    let state_before_sync = lagging_validator
+        .handle_chain_info_query(ChainInfoQuery::new(sender_chain))
+        .await?;
+    assert_eq!(state_before_sync.info.next_block_height, BlockHeight::ZERO);
+
+    // Synchronize the validator
+    let validator_address = net.validator_address(LAGGING_VALIDATOR_INDEX);
+    client
+        .sync_validator([&sender_chain], validator_address)
+        .await
+        .expect("Missing lagging validator name");
+
+    let state_after_sync = lagging_validator
+        .handle_chain_info_query(ChainInfoQuery::new(sender_chain))
+        .await?;
+    assert_eq!(
+        state_after_sync.info.next_block_height,
+        BlockHeight(BLOCKS_TO_CREATE as u64 + 1)
+    );
+
+    Ok(())
+}