Pin several large futures. (#4883)

Port of #4882.

## Motivation

On `testnet_conway`,
```
export LINERA_FAUCET_URL=https://faucet.testnet-conway.linera.net
cargo test test_wasm_end_to_end_fungible::remote_net_grpc --features remote-net
```
fails with a stack overflow.

## Proposal

`Box::pin` a lot of futures larger than 3 kB (using Claude, also to
resolve merge conflicts).

## Test Plan

CI

This change fixed it for me locally.

## Release Plan

- Noting to do.

## Links

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

Author: afck

linera-core/src/chain_worker/state.rs

@@ -1350,9 +1350,8 @@ where
         let (_, committee) = self.chain.current_committee()?;
         block.check_proposal_size(committee.policy().maximum_block_proposal_size)?;
 
-        let executed_block = self
-            .execute_block(&block, local_time, round, published_blobs)
-            .await?;
+        let executed_block =
+            Box::pin(self.execute_block(&block, local_time, round, published_blobs)).await?;
 
         // No need to sign: only used internally.
         let mut response = ChainInfoResponse::new(&self.chain, None);
@@ -1478,7 +1477,7 @@ where
         let block = if let Some(outcome) = outcome {
             outcome.clone().with(proposal.content.block.clone())
         } else {
-            self.execute_block(block, local_time, round.multi_leader(), &published_blobs)
+            Box::pin(self.execute_block(block, local_time, round.multi_leader(), &published_blobs))
                 .await?
         };
 
@@ -1613,10 +1612,12 @@ where
         let block_hash = CryptoHash::new(&block);
         self.execution_state_cache.insert_owned(
             &block_hash,
-            self.chain
-                .execution_state
-                .with_context(|ctx| InactiveContext(ctx.base_key().clone()))
-                .await,
+            Box::pin(
+                self.chain
+                    .execution_state
+                    .with_context(|ctx| InactiveContext(ctx.base_key().clone())),
+            )
+            .await,
         );
         Ok(block)
     }

linera-core/src/local_node.rs

@@ -89,7 +89,8 @@ where
         proposal: BlockProposal,
     ) -> Result<ChainInfoResponse, LocalNodeError> {
         // In local nodes, we can trust fully_handle_certificate to carry all actions eventually.
-        let (response, _actions) = self.node.state.handle_block_proposal(proposal).await?;
+        let (response, _actions) =
+            Box::pin(self.node.state.handle_block_proposal(proposal)).await?;
         Ok(response)
     }
 

linera-core/src/worker.rs

@@ -563,7 +563,7 @@ impl ProcessableCertificate for ConfirmedBlock {
         worker: &WorkerState<S>,
         certificate: ConfirmedBlockCertificate,
     ) -> Result<(ChainInfoResponse, NetworkActions), WorkerError> {
-        worker.handle_confirmed_certificate(certificate, None).await
+        Box::pin(worker.handle_confirmed_certificate(certificate, None)).await
     }
 }
 
@@ -572,7 +572,7 @@ impl ProcessableCertificate for ValidatedBlock {
         worker: &WorkerState<S>,
         certificate: ValidatedBlockCertificate,
     ) -> Result<(ChainInfoResponse, NetworkActions), WorkerError> {
-        worker.handle_validated_certificate(certificate).await
+        Box::pin(worker.handle_validated_certificate(certificate)).await
     }
 }
 
@@ -939,8 +939,13 @@ where
     ) -> Result<(ChainInfoResponse, NetworkActions), WorkerError> {
         match self.full_certificate(certificate).await? {
             Either::Left(confirmed) => {
-                self.handle_confirmed_certificate(confirmed, notify_when_messages_are_delivered)
-                    .await
+                Box::pin(
+                    self.handle_confirmed_certificate(
+                        confirmed,
+                        notify_when_messages_are_delivered,
+                    ),
+                )
+                .await
             }
             Either::Right(validated) => {
                 if let Some(notifier) = notify_when_messages_are_delivered {
@@ -949,7 +954,7 @@ where
                         warn!("Failed to notify message delivery to caller");
                     }
                 }
-                self.handle_validated_certificate(validated).await
+                Box::pin(self.handle_validated_certificate(validated)).await
             }
         }
     }
@@ -979,9 +984,9 @@ where
                 .collect::<Vec<_>>(),
         );
 
-        let (info, actions, _outcome) = self
-            .process_confirmed_block(certificate, notify_when_messages_are_delivered)
-            .await?;
+        let (info, actions, _outcome) =
+            Box::pin(self.process_confirmed_block(certificate, notify_when_messages_are_delivered))
+                .await?;
 
         #[cfg(with_metrics)]
         {
@@ -1030,7 +1035,7 @@ where
         #[cfg(with_metrics)]
         let cert_str = certificate.inner().to_log_str();
 
-        let (info, actions, _outcome) = self.process_validated_block(certificate).await?;
+        let (info, actions, _outcome) = Box::pin(self.process_validated_block(certificate)).await?;
         #[cfg(with_metrics)]
         {
             if matches!(_outcome, BlockOutcome::Processed) {

linera-execution/src/execution.rs

@@ -129,9 +129,11 @@ where
             &[],
         );
         txn_tracker.add_created_blob(blob);
-        ExecutionStateActor::new(self, &mut txn_tracker, &mut resource_controller)
-            .run_user_action(application_id, action, context.refund_grant_to(), None)
-            .await?;
+        Box::pin(
+            ExecutionStateActor::new(self, &mut txn_tracker, &mut resource_controller)
+                .run_user_action(application_id, action, context.refund_grant_to(), None),
+        )
+        .await?;
 
         Ok(())
     }

linera-views/src/backends/scylla_db.rs

@@ -350,9 +350,7 @@ impl ScyllaDbClient {
         let statement = self.get_multi_key_values_statement(map.len()).await?;
         let mut inputs = vec![root_key.to_vec()];
         inputs.extend(map.keys().cloned());
-        let mut rows = self
-            .session
-            .execute_iter(statement, &inputs)
+        let mut rows = Box::pin(self.session.execute_iter(statement, &inputs))
             .await?
             .rows_stream::<(Vec<u8>, Vec<u8>)>()?;
 
@@ -378,9 +376,7 @@ impl ScyllaDbClient {
         let statement = self.get_multi_keys_statement(map.len()).await?;
         let mut inputs = vec![root_key.to_vec()];
         inputs.extend(map.keys().cloned());
-        let mut rows = self
-            .session
-            .execute_iter(statement, &inputs)
+        let mut rows = Box::pin(self.session.execute_iter(statement, &inputs))
             .await?
             .rows_stream::<(Vec<u8>,)>()?;
 
@@ -471,13 +467,11 @@ impl ScyllaDbClient {
         let rows = match get_upper_bound_option(&key_prefix) {
             None => {
                 let values = (root_key.to_vec(), key_prefix.clone());
-                session
-                    .execute_iter(query_unbounded.clone(), values)
-                    .await?
+                Box::pin(session.execute_iter(query_unbounded.clone(), values)).await?
             }
             Some(upper_bound) => {
                 let values = (root_key.to_vec(), key_prefix.clone(), upper_bound);
-                session.execute_iter(query_bounded.clone(), values).await?
+                Box::pin(session.execute_iter(query_bounded.clone(), values)).await?
             }
         };
         let mut rows = rows.rows_stream::<(Vec<u8>,)>()?;
@@ -504,13 +498,11 @@ impl ScyllaDbClient {
         let rows = match get_upper_bound_option(&key_prefix) {
             None => {
                 let values = (root_key.to_vec(), key_prefix.clone());
-                session
-                    .execute_iter(query_unbounded.clone(), values)
-                    .await?
+                Box::pin(session.execute_iter(query_unbounded.clone(), values)).await?
             }
             Some(upper_bound) => {
                 let values = (root_key.to_vec(), key_prefix.clone(), upper_bound);
-                session.execute_iter(query_bounded.clone(), values).await?
+                Box::pin(session.execute_iter(query_bounded.clone(), values)).await?
             }
         };
         let mut rows = rows.rows_stream::<(Vec<u8>, Vec<u8>)>()?;
@@ -630,17 +622,13 @@ impl ReadableKeyValueStore for ScyllaDbStoreInternal {
     ) -> Result<Option<Vec<u8>>, ScyllaDbStoreInternalError> {
         let store = self.store.deref();
         let _guard = self.acquire().await;
-        store
-            .read_value_internal(&self.root_key, key.to_vec())
-            .await
+        Box::pin(store.read_value_internal(&self.root_key, key.to_vec())).await
     }
 
     async fn contains_key(&self, key: &[u8]) -> Result<bool, ScyllaDbStoreInternalError> {
         let store = self.store.deref();
         let _guard = self.acquire().await;
-        store
-            .contains_key_internal(&self.root_key, key.to_vec())
-            .await
+        Box::pin(store.contains_key_internal(&self.root_key, key.to_vec())).await
     }
 
     async fn contains_keys(
@@ -687,9 +675,7 @@ impl ReadableKeyValueStore for ScyllaDbStoreInternal {
     ) -> Result<Vec<Vec<u8>>, ScyllaDbStoreInternalError> {
         let store = self.store.deref();
         let _guard = self.acquire().await;
-        store
-            .find_keys_by_prefix_internal(&self.root_key, key_prefix.to_vec())
-            .await
+        Box::pin(store.find_keys_by_prefix_internal(&self.root_key, key_prefix.to_vec())).await
     }
 
     async fn find_key_values_by_prefix(
@@ -698,8 +684,7 @@ impl ReadableKeyValueStore for ScyllaDbStoreInternal {
     ) -> Result<Vec<(Vec<u8>, Vec<u8>)>, ScyllaDbStoreInternalError> {
         let store = self.store.deref();
         let _guard = self.acquire().await;
-        store
-            .find_key_values_by_prefix_internal(&self.root_key, key_prefix.to_vec())
+        Box::pin(store.find_key_values_by_prefix_internal(&self.root_key, key_prefix.to_vec()))
             .await
     }
 }
@@ -792,7 +777,7 @@ impl KeyValueDatabase for ScyllaDbDatabaseInternal {
         let statement = session
             .prepare(format!("DESCRIBE KEYSPACE {}", KEYSPACE))
             .await?;
-        let result = session.execute_iter(statement, &[]).await;
+        let result = Box::pin(session.execute_iter(statement, &[])).await;
         let miss_msg = format!("'{}' not found in keyspaces", KEYSPACE);
         let result = match result {
             Ok(result) => result,
@@ -832,7 +817,7 @@ impl KeyValueDatabase for ScyllaDbDatabaseInternal {
             .await?;
 
         // Execute the query
-        let rows = self.store.session.execute_iter(statement, &[]).await?;
+        let rows = Box::pin(self.store.session.execute_iter(statement, &[])).await?;
         let mut rows = rows.rows_stream::<(Vec<u8>,)>()?;
         let mut root_keys = BTreeSet::new();
         while let Some(row) = rows.next().await {