[stateless_validation] Broadcast chunk endrosement to next 5 block pr…

…oducers (#10449)

In the current implementation, we only send the chunk endorsements to
the next block producer. As discussed in [this Zulip
thread](https://near.zulipchat.com/#narrow/stream/407237-pagoda.2Fcore.2Fstateless-validation/topic/chunk.20endorsements.20with.20missing.20chunks.2Fblocks/near/416015998),
it's possible for a chunk to be included in a block at a height
different from that it was created for. This leads to the situation
where this new block producer hasn't received the chunk endorsements for
this chunk.

The solution here is to broadcast the chunk endorsement message to the
next 5 block producers and hope that the chunk would get included in any
one of them in cases of skipped blocks.

Note that this also means we are always going to use the
chunk_validators as of the `chunk.height_created` and NOT
`chunk.height_included`

chain/client/src/chunk_validation.rs

@@ -1,3 +1,4 @@
+use itertools::Itertools;
 use near_async::messaging::{CanSend, Sender};
 use near_chain::chain::{
     apply_new_chunk, apply_old_chunk, NewChunkData, NewChunkResult, OldChunkData, OldChunkResult,
@@ -32,6 +33,13 @@ use crate::Client;
 // Ideally, we should not be processing more than num_shards chunks at a time.
 const NUM_CHUNK_ENDORSEMENTS_CACHE_COUNT: usize = 100;
 
+// After validating a chunk state witness, we ideally need to send the chunk endorsement
+// to just the next block producer at height h. However, it's possible that blocks at height
+// h may be skipped and block producer at height h+1 picks up the chunk. We need to ensure
+// that these later block producers also receive the chunk endorsement.
+// Keeping a threshold of 5 block producers should be sufficient for most scenarios.
+const NUM_NEXT_BLOCK_PRODUCERS_TO_SEND_CHUNK_ENDORSEMENT: u64 = 5;
+
 /// A module that handles chunk validation logic. Chunk validation refers to a
 /// critical process of stateless validation, where chunk validators (certain
 /// validators selected to validate the chunk) verify that the chunk's state
@@ -98,8 +106,13 @@ impl ChunkValidator {
             self.epoch_manager.as_ref(),
         )?;
 
-        let block_producer =
-            self.epoch_manager.get_block_producer(&epoch_id, chunk_header.height_created())?;
+        // Send the chunk endorsement to the next NUM_NEXT_BLOCK_PRODUCERS_TO_SEND_CHUNK_ENDORSEMENT block producers.
+        // It's possible we may reach the end of the epoch, in which case, ignore the error from get_block_producer.
+        let block_height = chunk_header.height_created();
+        let block_producers = (0..NUM_NEXT_BLOCK_PRODUCERS_TO_SEND_CHUNK_ENDORSEMENT)
+            .map_while(|i| self.epoch_manager.get_block_producer(&epoch_id, block_height + i).ok())
+            .collect_vec();
+        assert!(!block_producers.is_empty());
 
         let network_sender = self.network_sender.clone();
         let signer = self.my_signer.clone().unwrap();
@@ -116,13 +129,15 @@ impl ChunkValidator {
                     tracing::debug!(
                         target: "chunk_validation",
                         chunk_hash=?chunk_header.chunk_hash(),
-                        block_producer=%block_producer,
+                        ?block_producers,
                         "Chunk validated successfully, sending endorsement",
                     );
                     let endorsement = ChunkEndorsement::new(chunk_header.chunk_hash(), signer);
-                    network_sender.send(PeerManagerMessageRequest::NetworkRequests(
-                        NetworkRequests::ChunkEndorsement(block_producer, endorsement),
-                    ));
+                    for block_producer in block_producers {
+                        network_sender.send(PeerManagerMessageRequest::NetworkRequests(
+                            NetworkRequests::ChunkEndorsement(block_producer, endorsement.clone()),
+                        ));
+                    }
                 }
                 Err(err) => {
                     tracing::error!("Failed to validate chunk: {:?}", err);
@@ -542,23 +557,6 @@ impl Client {
     ) -> Result<(), Error> {
         let chunk_hash = endorsement.chunk_hash();
         let account_id = &endorsement.account_id;
-        let chunk_header = self.chain.get_chunk(chunk_hash)?.cloned_header();
-
-        // Note that we are using the chunk_header.height_created param here to determine the block_producer
-        // This only works when height created for a chunk is the same as the height_included during block production
-        let epoch_id =
-            self.epoch_manager.get_epoch_id_from_prev_block(chunk_header.prev_block_hash())?;
-        let block_producer =
-            self.epoch_manager.get_block_producer(&epoch_id, chunk_header.height_created())?;
-        // Verify that we are the block producer for this height.
-        if !self
-            .validator_signer
-            .as_ref()
-            .is_some_and(|my_signer| my_signer.validator_id() == &block_producer)
-        {
-            tracing::error!(target: "chunk_validation", ?endorsement, ?block_producer, "Received chunk endorsement for non-block producer.");
-            return Err(Error::InvalidChunkEndorsement);
-        }
 
         // If we have already processed this chunk endorsement, return early.
         if self
@@ -571,13 +569,18 @@ impl Client {
             return Ok(());
         }
 
+        let chunk_header = self.chain.get_chunk(chunk_hash)?.cloned_header();
         if !self.epoch_manager.verify_chunk_endorsement(&chunk_header, &endorsement)? {
             tracing::error!(target: "chunk_validation", ?endorsement, "Invalid chunk endorsement.");
             return Err(Error::InvalidChunkEndorsement);
         }
 
         // If we are the current block producer, we store the chunk endorsement for each chunk which
         // would later be used during block production to check whether to include the chunk or not.
+        // TODO(stateless_validation): It's possible for a malicious validator to send endorsements
+        // for 100 unique chunks thus pushing out current valid endorsements from our cache.
+        // Maybe add check to ensure we don't accept endorsements from chunks already included in some block?
+        // Maybe add check to ensure we don't accept endorsements from chunks that have too old height_created?
         tracing::debug!(target: "chunk_validation", ?endorsement, "Received and saved chunk endorsement.");
         self.chunk_validator
             .chunk_endorsements