Reduce size of futures in HTTP API to prevent stack overflows (#5104)

* Box::pin a few big futures

* Arc the blocks early in publication

* Fix more tests

beacon_node/beacon_chain/tests/block_verification.rs

@@ -1142,11 +1142,7 @@ async fn verify_block_for_gossip_slashing_detection() {
     let ((block1, blobs1), _) = harness.make_block(state.clone(), Slot::new(1)).await;
     let ((block2, _blobs2), _) = harness.make_block(state, Slot::new(1)).await;
 
-    let verified_block = harness
-        .chain
-        .verify_block_for_gossip(Arc::new(block1))
-        .await
-        .unwrap();
+    let verified_block = harness.chain.verify_block_for_gossip(block1).await.unwrap();
 
     if let Some((kzg_proofs, blobs)) = blobs1 {
         let sidecars =
@@ -1174,12 +1170,7 @@ async fn verify_block_for_gossip_slashing_detection() {
         )
         .await
         .unwrap();
-    unwrap_err(
-        harness
-            .chain
-            .verify_block_for_gossip(Arc::new(block2))
-            .await,
-    );
+    unwrap_err(harness.chain.verify_block_for_gossip(block2).await);
 
     // Slasher should have been handed the two conflicting blocks and crafted a slashing.
     slasher.process_queued(Epoch::new(0)).unwrap();
@@ -1198,11 +1189,7 @@ async fn verify_block_for_gossip_doppelganger_detection() {
     let state = harness.get_current_state();
     let ((block, _), _) = harness.make_block(state.clone(), Slot::new(1)).await;
 
-    let verified_block = harness
-        .chain
-        .verify_block_for_gossip(Arc::new(block))
-        .await
-        .unwrap();
+    let verified_block = harness.chain.verify_block_for_gossip(block).await.unwrap();
     let attestations = verified_block.block.message().body().attestations().clone();
     harness
         .chain
@@ -1564,7 +1551,6 @@ async fn import_duplicate_block_unrealized_justification() {
     let slot = harness.get_current_slot();
     let (block_contents, _) = harness.make_block(state.clone(), slot).await;
     let (block, _) = block_contents;
-    let block = Arc::new(block);
     let block_root = block.canonical_root();
 
     // Create two verified variants of the block, representing the same block being processed in

beacon_node/beacon_chain/tests/payload_invalidation.rs

@@ -319,7 +319,7 @@ impl InvalidPayloadRig {
                         .get_full_block(&block_root)
                         .unwrap()
                         .unwrap(),
-                    block,
+                    *block,
                     "block from db must match block imported"
                 );
             }
@@ -700,7 +700,7 @@ async fn invalidates_all_descendants() {
         .chain
         .process_block(
             fork_block.canonical_root(),
-            Arc::new(fork_block),
+            fork_block,
             NotifyExecutionLayer::Yes,
             || Ok(()),
         )
@@ -800,7 +800,7 @@ async fn switches_heads() {
         .chain
         .process_block(
             fork_block.canonical_root(),
-            Arc::new(fork_block),
+            fork_block,
             NotifyExecutionLayer::Yes,
             || Ok(()),
         )
@@ -1044,8 +1044,7 @@ async fn invalid_parent() {
     // Produce another block atop the parent, but don't import yet.
     let slot = parent_block.slot() + 1;
     rig.harness.set_current_slot(slot);
-    let (block_tuple, state) = rig.harness.make_block(parent_state, slot).await;
-    let block = Arc::new(block_tuple.0);
+    let ((block, _), state) = rig.harness.make_block(parent_state, slot).await;
     let block_root = block.canonical_root();
     assert_eq!(block.parent_root(), parent_root);
 
@@ -1865,7 +1864,7 @@ impl InvalidHeadSetup {
                     .state_at_slot(slot - 1, StateSkipConfig::WithStateRoots)
                     .unwrap();
                 let (fork_block_tuple, _) = rig.harness.make_block(parent_state, slot).await;
-                opt_fork_block = Some(Arc::new(fork_block_tuple.0));
+                opt_fork_block = Some(fork_block_tuple.0);
             } else {
                 // Skipped slot.
             };

beacon_node/beacon_chain/tests/store_tests.rs

@@ -2268,17 +2268,17 @@ async fn garbage_collect_temp_states_from_failed_block() {
         let block_slot = Slot::new(2 * slots_per_epoch);
         let ((signed_block, _), state) = harness.make_block(genesis_state, block_slot).await;
 
-        let (mut block, _) = signed_block.deconstruct();
+        let (mut block, _) = (*signed_block).clone().deconstruct();
 
         // Mutate the block to make it invalid, and re-sign it.
         *block.state_root_mut() = Hash256::repeat_byte(0xff);
         let proposer_index = block.proposer_index() as usize;
-        let block = block.sign(
+        let block = Arc::new(block.sign(
             &harness.validator_keypairs[proposer_index].sk,
             &state.fork(),
             state.genesis_validators_root(),
             &harness.spec,
-        );
+        ));
 
         // The block should be rejected, but should store a bunch of temporary states.
         harness.set_current_slot(block_slot);
@@ -2677,7 +2677,7 @@ async fn process_blocks_and_attestations_for_unaligned_checkpoint() {
         .chain
         .process_block(
             invalid_fork_block.canonical_root(),
-            Arc::new(invalid_fork_block.clone()),
+            invalid_fork_block.clone(),
             NotifyExecutionLayer::Yes,
             || Ok(()),
         )
@@ -2690,7 +2690,7 @@ async fn process_blocks_and_attestations_for_unaligned_checkpoint() {
         .chain
         .process_block(
             valid_fork_block.canonical_root(),
-            Arc::new(valid_fork_block.clone()),
+            valid_fork_block.clone(),
             NotifyExecutionLayer::Yes,
             || Ok(()),
         )