Don't do heavy syncing of sender chains unless requested. (#4706) (#4720)

Backport of #4706.

Merge conflicts:
* `linera-client/src/benchmark.rs`
* `linera-service/src/cli_wrappers/wallet.rs`

## Motivation

`find_received_certificates_from_validator` and
`find_received_certificates` can take a lot of time (and memory) if
there are a lot of incoming messages.

## Proposal

Only do them as part of `synchronize_from_validators`. On notification,
`prepare_chain` and when we're about to propose a block, only download
what's necessary.

Download and process sender certificates in batches, to avoid the high
memory usage.

Download them in a background task in the node service, with some sleep
time in between, to limit interference with the other node service
tasks.

(Mostly written with Claude Sonnet 4.5.)

## Test Plan

CI
A new test was added for the updated `prepare_chain`.
`test_sparse_sender_chain` already exists, and tests notification
handling with a sparse chain.

## Release Plan

- These changes should be released in a new SDK.

## Links

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

Author: afck

CLI.md

@@ -729,6 +729,9 @@ Run a GraphQL service to explore and extend the chains of the wallet
 
   Default value: `0`
 * `--port <PORT>` — The port on which to run the server
+* `--sync-sleep-ms <SYNC_SLEEP_MS>` — Milliseconds to sleep between batches during background certificate synchronization
+
+  Default value: `500`
 
 
 

linera-chain/src/chain.rs

@@ -539,6 +539,33 @@ where
         Ok(())
     }
 
+    /// Collects all missing sender blocks from removed bundles across all inboxes.
+    /// Returns a map of origin chain IDs to their respective missing block heights.
+    #[instrument(target = "telemetry_only", skip_all, fields(
+        chain_id = %self.chain_id()
+    ))]
+    pub async fn collect_missing_sender_blocks(
+        &self,
+    ) -> Result<BTreeMap<ChainId, Vec<BlockHeight>>, ChainError> {
+        let pairs = self.inboxes.try_load_all_entries().await?;
+        let max_stream_queries = self.context().store().max_stream_queries();
+        let stream = stream::iter(pairs)
+            .map(|(origin, inbox)| async move {
+                let mut missing_heights = Vec::new();
+                let bundles = inbox.removed_bundles.elements().await?;
+                for bundle in bundles {
+                    missing_heights.push(bundle.height);
+                }
+                Ok::<(ChainId, Vec<BlockHeight>), ChainError>((origin, missing_heights))
+            })
+            .buffer_unordered(max_stream_queries);
+        let results: Vec<(ChainId, Vec<BlockHeight>)> = stream.try_collect().await?;
+        Ok(results
+            .into_iter()
+            .filter(|(_, heights)| !heights.is_empty())
+            .collect())
+    }
+
     pub async fn next_block_height_to_receive(
         &self,
         origin: &ChainId,

linera-chain/src/lib.rs

@@ -56,7 +56,7 @@ pub enum ChainError {
         height: BlockHeight,
     },
     #[error(
-        "Message in block proposed to {chain_id:?} does not match the previously received messages from \
+        "Message in block proposed to {chain_id} does not match the previously received messages from \
         origin {origin:?}: was {bundle:?} instead of {previous_bundle:?}"
     )]
     UnexpectedMessage {
@@ -66,7 +66,7 @@ pub enum ChainError {
         previous_bundle: Box<MessageBundle>,
     },
     #[error(
-        "Message in block proposed to {chain_id:?} is out of order compared to previous messages \
+        "Message in block proposed to {chain_id} is out of order compared to previous messages \
          from origin {origin:?}: {bundle:?}. Block and height should be at least: \
          {next_height}, {next_index}"
     )]
@@ -78,7 +78,7 @@ pub enum ChainError {
         next_index: u32,
     },
     #[error(
-        "Block proposed to {chain_id:?} is attempting to reject protected message \
+        "Block proposed to {chain_id} is attempting to reject protected message \
         {posted_message:?}"
     )]
     CannotRejectMessage {
@@ -87,7 +87,7 @@ pub enum ChainError {
         posted_message: Box<PostedMessage>,
     },
     #[error(
-        "Block proposed to {chain_id:?} is attempting to skip a message bundle \
+        "Block proposed to {chain_id} is attempting to skip a message bundle \
          that cannot be skipped: {bundle:?}"
     )]
     CannotSkipMessage {
@@ -96,7 +96,7 @@ pub enum ChainError {
         bundle: Box<MessageBundle>,
     },
     #[error(
-        "Incoming message bundle in block proposed to {chain_id:?} has timestamp \
+        "Incoming message bundle in block proposed to {chain_id} has timestamp \
         {bundle_timestamp:}, which is later than the block timestamp {block_timestamp:}."
     )]
     IncorrectBundleTimestamp {

linera-client/src/benchmark.rs

@@ -126,7 +126,9 @@ impl<Env: Environment> Benchmark<Env> {
         let notifier = Arc::new(Notify::new());
         let barrier = Arc::new(Barrier::new(num_chains + 1));
 
-        let chain_listener_result = chain_listener.run().await;
+        let chain_listener_result = chain_listener
+            .run(Some(500)) // Default sync_sleep_ms for benchmarks
+            .await;
 
         let chain_listener_handle =
             tokio::spawn(async move { chain_listener_result?.await }.in_current_span());

linera-client/src/chain_listener.rs

@@ -182,7 +182,7 @@ pub struct ChainListener<C: ClientContext> {
     cancellation_token: CancellationToken,
 }
 
-impl<C: ClientContext> ChainListener<C> {
+impl<C: ClientContext + 'static> ChainListener<C> {
     /// Creates a new chain listener given client chains.
     pub fn new(
         config: ChainListenerConfig,
@@ -202,13 +202,16 @@ impl<C: ClientContext> ChainListener<C> {
 
     /// Runs the chain listener.
     #[instrument(skip(self))]
-    pub async fn run(mut self) -> Result<impl Future<Output = Result<(), Error>>, Error> {
+    pub async fn run(
+        mut self,
+        sync_sleep_ms: Option<u64>,
+    ) -> Result<impl Future<Output = Result<(), Error>>, Error> {
         let chain_ids = {
             let guard = self.context.lock().await;
             let admin_chain_id = guard.wallet().genesis_admin_chain();
             guard
                 .make_chain_client(admin_chain_id)
-                .synchronize_from_validators()
+                .synchronize_chain_state(admin_chain_id)
                 .await?;
             BTreeSet::from_iter(
                 guard
@@ -219,6 +222,30 @@ impl<C: ClientContext> ChainListener<C> {
             )
         };
 
+        // Start background tasks to sync received certificates for each chain,
+        // if enabled (sync_sleep_ms is Some).
+        if let Some(sync_sleep_ms) = sync_sleep_ms {
+            let context = Arc::clone(&self.context);
+            let cancellation_token = self.cancellation_token.clone();
+            for chain_id in &chain_ids {
+                let context = Arc::clone(&context);
+                let cancellation_token = cancellation_token.clone();
+                let chain_id = *chain_id;
+                drop(linera_base::task::spawn(async move {
+                    if let Err(e) = Self::background_sync_received_certificates(
+                        context,
+                        sync_sleep_ms,
+                        chain_id,
+                        cancellation_token,
+                    )
+                    .await
+                    {
+                        warn!("Background sync failed for chain {chain_id}: {e}");
+                    }
+                }));
+            }
+        }
+
         Ok(async {
             self.listen_recursively(chain_ids).await?;
             loop {
@@ -333,6 +360,46 @@ impl<C: ClientContext> ChainListener<C> {
         Ok(())
     }
 
+    /// Background task that syncs received certificates in small batches.
+    /// This discovers unacknowledged sender blocks gradually without overwhelming the system.
+    #[instrument(skip(context, cancellation_token))]
+    async fn background_sync_received_certificates(
+        context: Arc<Mutex<C>>,
+        sync_sleep_ms: u64,
+        chain_id: ChainId,
+        cancellation_token: CancellationToken,
+    ) -> Result<(), Error> {
+        info!("Starting background certificate sync for chain {chain_id}");
+        let client = context.lock().await.make_chain_client(chain_id);
+
+        loop {
+            // Check if we should stop.
+            if cancellation_token.is_cancelled() {
+                info!("Background sync cancelled for chain {chain_id}");
+                break;
+            }
+
+            // Sleep to avoid overwhelming the system.
+            Self::sleep(sync_sleep_ms).await;
+
+            // Sync one batch.
+            match client.sync_received_certificates_batch().await {
+                Ok(has_more) => {
+                    if !has_more {
+                        info!("Background sync completed for chain {chain_id}");
+                        break;
+                    }
+                }
+                Err(e) => {
+                    warn!("Error syncing batch for chain {chain_id}: {e}. Will retry.");
+                    // Continue trying despite errors - validators might be temporarily unavailable.
+                }
+            }
+        }
+
+        Ok(())
+    }
+
     /// Starts listening for notifications about the given chain.
     ///
     /// Returns all publishing chains, that we also need to listen to.

linera-client/src/unit_tests/chain_listener.rs

@@ -148,7 +148,7 @@ async fn test_chain_listener() -> anyhow::Result<()> {
     let cancellation_token = CancellationToken::new();
     let child_token = cancellation_token.child_token();
     let chain_listener = ChainListener::new(config, context, storage, child_token)
-        .run()
+        .run(None) // Unit test doesn't need background sync
         .await
         .unwrap();
 
@@ -210,7 +210,7 @@ async fn test_chain_listener_admin_chain() -> anyhow::Result<()> {
     let cancellation_token = CancellationToken::new();
     let child_token = cancellation_token.child_token();
     let chain_listener = ChainListener::new(config, context, storage.clone(), child_token)
-        .run()
+        .run(None) // Unit test doesn't need background sync
         .await
         .unwrap();