Synchronize chain state after notification stream is established. (#4361)

## Motivation

`test_wasm_end_to_end_social_event_streams::remote_net_grpc` is failing
on `devnet-2025-08-15`. The problem seems to be that the relevant block
on the publisher chain arrives _after_ we synchronize that chain
explicitly but _before_ the notification streams from the validators are
established, so we miss that block.

It's difficult to reproduce this issue in another way; locally, the
tests all pass.

## Proposal

Synchronize after establishing the notification stream, for each
validator.

## Test Plan

For me, this fixes the test.

## Release Plan

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

## Links

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

Author: afck

linera-core/src/client/mod.rs

@@ -1854,10 +1854,16 @@ impl<Env: Environment> ChainClient<Env> {
 
     /// Obtains the committee for the current epoch of the local chain.
     #[instrument(level = "trace")]
-    pub async fn local_committee(&self) -> Result<Committee, LocalNodeError> {
-        self.chain_info_with_committees()
-            .await?
-            .into_current_committee()
+    pub async fn local_committee(&self) -> Result<Committee, ChainClientError> {
+        let info = match self.chain_info_with_committees().await {
+            Ok(info) => info,
+            Err(LocalNodeError::BlobsNotFound(_)) => {
+                self.synchronize_chain_state(self.chain_id).await?;
+                self.chain_info_with_committees().await?
+            }
+            Err(err) => return Err(err.into()),
+        };
+        Ok(info.into_current_committee()?)
     }
 
     /// Obtains the committee for the latest epoch on the admin chain.
@@ -3707,14 +3713,6 @@ impl<Env: Environment> ChainClient<Env> {
         let mut senders = HashMap::new(); // Senders to cancel notification streams.
         let notifications = self.subscribe()?;
         let (abortable_notifications, abort) = stream::abortable(self.subscribe()?);
-        let sync_result = if self.is_tracked() {
-            self.synchronize_from_validators().await
-        } else {
-            self.synchronize_chain_state(self.chain_id).await
-        };
-        if let Err(error) = sync_result {
-            error!("Failed to synchronize from validators: {}", error);
-        }
 
         // Beware: if this future ceases to make progress, notification processing will
         // deadlock, because of the issue described in
@@ -3738,10 +3736,10 @@ impl<Env: Environment> ChainClient<Env> {
                     .await
             {
                 if let Reason::NewBlock { .. } = notification.reason {
-                    match await_while_polling(
+                    match Box::pin(await_while_polling(
                         this.update_notification_streams(&mut senders).fuse(),
                         &mut process_notifications,
-                    )
+                    ))
                     .await
                     {
                         Ok(handler) => process_notifications.push(handler),
@@ -3766,14 +3764,14 @@ impl<Env: Environment> ChainClient<Env> {
         &self,
         senders: &mut HashMap<ValidatorPublicKey, AbortHandle>,
     ) -> Result<impl Future<Output = ()>, ChainClientError> {
-        let (chain_id, nodes, local_node) = {
+        let (nodes, local_node) = {
             let committee = self.local_committee().await?;
             let nodes: HashMap<_, _> = self
                 .client
                 .validator_node_provider()
                 .make_nodes(&committee)?
                 .collect();
-            (self.chain_id, nodes, self.client.local_node.clone())
+            (nodes, self.client.local_node.clone())
         };
         // Drop removed validators.
         senders.retain(|validator, abort| {
@@ -3788,9 +3786,19 @@ impl<Env: Environment> ChainClient<Env> {
             let hash_map::Entry::Vacant(entry) = senders.entry(public_key) else {
                 continue;
             };
+            let this = self.clone();
             let stream = stream::once({
                 let node = node.clone();
-                async move { node.subscribe(vec![chain_id]).await }
+                async move {
+                    let stream = node.subscribe(vec![this.chain_id]).await?;
+                    // Only now the notification stream is established. We may have missed
+                    // notifications since the last time we synchronized.
+                    let remote_node = RemoteNode { public_key, node };
+                    this.client
+                        .synchronize_chain_state_from(&remote_node, this.chain_id)
+                        .await?;
+                    Ok::<_, ChainClientError>(stream)
+                }
             })
             .filter_map(move |result| async move {
                 if let Err(error) = &result {

linera-service/tests/linera_net_tests.rs

@@ -1873,7 +1873,6 @@ async fn test_wasm_end_to_end_counter_publish_create(config: impl LineraNetConfi
 #[cfg_attr(feature = "kubernetes", test_case(SharedLocalKubernetesNetTestingConfig::new(Network::Grpc, BuildArg::Build) ; "kubernetes_grpc"))]
 #[cfg_attr(feature = "remote-net", test_case(RemoteNetTestingConfig::new(None) ; "remote_net_grpc"))]
 #[test_log::test(tokio::test)]
-#[ignore] // TODO(#4363)
 async fn test_wasm_end_to_end_social_event_streams(config: impl LineraNetConfig) -> Result<()> {
     use linera_base::time::Instant;
     use social::SocialAbi;

linera-service/tests/local_net_tests.rs

@@ -145,7 +145,9 @@ async fn test_end_to_end_reconfiguration(config: LocalNetConfig) -> Result<()> {
     client.query_validators(Some(chain_1)).await?;
 
     if matches!(network, Network::Grpc) {
-        assert_eq!(faucet.current_validators().await?.len(), 5);
+        assert!(
+            eventually(|| async { faucet.current_validators().await.unwrap().len() == 5 }).await
+        );
     }
 
     // Add 6th validator