Two-phase chain listener (#4436)

## Motivation

Currently we don't synchronize the chains when starting up the client.
Now that we also don't pessimistically synchronize chains before
performing operations, we sometimes run into the situation where we want
to do something with a chain whose blocks haven't been seen yet (e.g.
because it came from the faucet), which leads to an error.

## Proposal

Ideally when we find a block from the future we should synchronize the
offending chain. But for now it should be sufficient to make sure to
synchronize the initial chains at startup.

Split `ChainListener::run` into two futures: the latter is the full
chain listener loop, as before, but first there's an initialization step
that synchronizes all the initial chains, and the caller can wait on
this in serial to ensure the chains are synchronized before performing
operations on the client.

## Test Plan

CI should check for regressions; it would be nice to add a test for this
case, perhaps after testnet release.

## Release Plan

- These changes should be backported to the latest `devnet` branch
- 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: Twey

linera-client/src/benchmark.rs

@@ -130,14 +130,10 @@ impl<Env: Environment> Benchmark<Env> {
         let notifier = Arc::new(Notify::new());
         let barrier = Arc::new(Barrier::new(num_chains + 1));
 
-        let chain_listener_handle = tokio::spawn(
-            async move {
-                if let Err(e) = chain_listener.run().await {
-                    warn!("Chain listener error: {}", e);
-                }
-            }
-            .instrument(tracing::info_span!("chain_listener")),
-        );
+        let chain_listener_result = chain_listener.run().await;
+
+        let chain_listener_handle =
+            tokio::spawn(async move { chain_listener_result?.await }.in_current_span());
 
         let bps_control_task = Self::bps_control_task(
             &barrier,
@@ -216,7 +212,10 @@ impl<Env: Environment> Benchmark<Env> {
         if let Some(runtime_control_task) = runtime_control_task {
             runtime_control_task.await?;
         }
-        chain_listener_handle.await?;
+
+        if let Err(e) = chain_listener_handle.await? {
+            tracing::error!("chain listener error: {e}");
+        }
 
         Ok(())
     }

linera-client/src/chain_listener.rs

@@ -10,7 +10,7 @@ use std::{
 use futures::{
     future::{join_all, select_all},
     lock::Mutex,
-    FutureExt as _, StreamExt,
+    Future, FutureExt as _, StreamExt,
 };
 use linera_base::{
     crypto::{CryptoHash, Signer},
@@ -202,25 +202,37 @@ impl<C: ClientContext> ChainListener<C> {
 
     /// Runs the chain listener.
     #[instrument(skip(self))]
-    pub async fn run(mut self) -> Result<(), Error> {
+    pub async fn run(mut self) -> Result<impl Future<Output = Result<(), Error>>, Error> {
         let chain_ids = {
             let guard = self.context.lock().await;
-            let mut chain_ids = BTreeSet::from_iter(guard.wallet().chain_ids());
-            chain_ids.insert(guard.wallet().genesis_admin_chain());
-            chain_ids
+            let admin_chain_id = guard.wallet().genesis_admin_chain();
+            guard
+                .make_chain_client(admin_chain_id)
+                .synchronize_from_validators()
+                .await?;
+            BTreeSet::from_iter(
+                guard
+                    .wallet()
+                    .chain_ids()
+                    .into_iter()
+                    .chain([admin_chain_id]),
+            )
         };
-        self.listen_recursively(chain_ids).await?;
-        loop {
-            match self.next_action().await? {
-                Action::ProcessInbox(chain_id) => self.maybe_process_inbox(chain_id).await?,
-                Action::Notification(notification) => {
-                    self.process_notification(notification).await?
+
+        Ok(async {
+            self.listen_recursively(chain_ids).await?;
+            loop {
+                match self.next_action().await? {
+                    Action::ProcessInbox(chain_id) => self.maybe_process_inbox(chain_id).await?,
+                    Action::Notification(notification) => {
+                        self.process_notification(notification).await?
+                    }
+                    Action::Stop => break,
                 }
-                Action::Stop => break,
             }
-        }
-        join_all(self.listening.into_values().map(|client| client.stop())).await;
-        Ok(())
+            join_all(self.listening.into_values().map(|client| client.stop())).await;
+            Ok(())
+        })
     }
 
     /// Processes a notification, updating local chains and validators as needed.
@@ -317,6 +329,7 @@ impl<C: ClientContext> ChainListener<C> {
         while let Some(chain_id) = chain_ids.pop_first() {
             chain_ids.extend(self.listen(chain_id).await?);
         }
+
         Ok(())
     }
 

linera-client/src/unit_tests/chain_listener.rs

@@ -148,12 +148,12 @@ async fn test_chain_listener() -> anyhow::Result<()> {
     let context = Arc::new(Mutex::new(context));
     let cancellation_token = CancellationToken::new();
     let child_token = cancellation_token.child_token();
-    let handle = linera_base::task::spawn(async move {
-        ChainListener::new(config, context, storage, child_token)
-            .run()
-            .await
-            .unwrap()
-    });
+    let chain_listener = ChainListener::new(config, context, storage, child_token)
+        .run()
+        .await
+        .unwrap();
+
+    let handle = linera_base::task::spawn(async move { chain_listener.await.unwrap() });
     // Transfer one token to chain 0. The listener should eventually become leader and receive
     // the message.
     let recipient0 = Account::chain(chain_id0);
@@ -209,15 +209,12 @@ async fn test_chain_listener_admin_chain() -> anyhow::Result<()> {
     let context = Arc::new(Mutex::new(context));
     let cancellation_token = CancellationToken::new();
     let child_token = cancellation_token.child_token();
-    let handle = linera_base::task::spawn({
-        let storage = storage.clone();
-        async move {
-            ChainListener::new(config, context, storage, child_token)
-                .run()
-                .await
-                .unwrap()
-        }
-    });
+    let chain_listener = ChainListener::new(config, context, storage.clone(), child_token)
+        .run()
+        .await
+        .unwrap();
+
+    let handle = linera_base::task::spawn(async move { chain_listener.await.unwrap() });
     // Burn one token.
     let certificate = client0
         .burn(AccountOwner::CHAIN, Amount::ONE)

linera-faucet/server/src/lib.rs

@@ -682,7 +682,8 @@ where
             self.storage,
             cancellation_token.clone(),
         )
-        .run();
+        .run()
+        .await?;
         let batch_processor_task = batch_processor.run(cancellation_token.clone());
         let tcp_listener =
             tokio::net::TcpListener::bind(SocketAddr::from(([0, 0, 0, 0], port))).await?;

linera-service/src/node_service.rs

@@ -895,7 +895,9 @@ where
         let storage = self.context.lock().await.storage().clone();
 
         let chain_listener =
-            ChainListener::new(self.config, self.context, storage, cancellation_token).run();
+            ChainListener::new(self.config, self.context, storage, cancellation_token)
+                .run()
+                .await?;
         let mut chain_listener = Box::pin(chain_listener).fuse();
         let tcp_listener =
             tokio::net::TcpListener::bind(SocketAddr::from(([0, 0, 0, 0], port))).await?;

linera-web/src/lib.rs

@@ -8,6 +8,7 @@ This module defines the client API for the Web extension.
 // We sometimes need functions in this module to be async in order to
 // ensure the generated code will return a `Promise`.
 #![allow(clippy::unused_async)]
+#![recursion_limit = "256"]
 
 pub mod signer;
 
@@ -211,20 +212,24 @@ impl Client {
             signer,
         )));
         let client_context_clone = client_context.clone();
-        wasm_bindgen_futures::spawn_local(async move {
-            if let Err(error) = ChainListener::new(
-                ChainListenerConfig::default(),
-                client_context_clone,
-                storage,
-                tokio_util::sync::CancellationToken::new(),
-            )
-            .run()
-            .boxed_local()
-            .await
-            {
-                tracing::error!("ChainListener error: {error:?}");
+        let chain_listener = ChainListener::new(
+            ChainListenerConfig::default(),
+            client_context_clone,
+            storage,
+            tokio_util::sync::CancellationToken::new(),
+        )
+        .run()
+        .boxed_local()
+        .await?
+        .boxed_local();
+        wasm_bindgen_futures::spawn_local(
+            async move {
+                if let Err(error) = chain_listener.await {
+                    tracing::error!("ChainListener error: {error:?}");
+                }
             }
-        });
+            .boxed_local(),
+        );
         log::info!("Linera Web client successfully initialized");
         Ok(Self { client_context })
     }