feat(contract distribution) Exclude chunk producers from target of co…

…ntract accesses message (#12266)

Since the chunk producers will already have the contract code, skip
sending the message to those producers.

For this we introduce a new method to EpochManager to get the chunk
producers for a given shard, and take the difference of chunk-validators
from this set.

chain/chain/src/test_utils/kv_runtime.rs

@@ -738,6 +738,18 @@ impl EpochManagerAdapter for MockEpochManager {
         Ok(vec![])
     }
 
+    fn get_epoch_chunk_producers_for_shard(
+        &self,
+        epoch_id: &EpochId,
+        shard_id: ShardId,
+    ) -> Result<Vec<AccountId>, EpochError> {
+        let valset = self.get_valset_for_epoch(epoch_id)?;
+        let shard_layout = self.get_shard_layout(epoch_id)?;
+        let shard_index = shard_layout.get_shard_index(shard_id);
+        let chunk_producers = self.get_chunk_producers(valset, shard_index);
+        Ok(chunk_producers.into_iter().map(|vs| vs.take_account_id()).collect())
+    }
+
     /// We need to override the default implementation to make
     /// `Chain::should_produce_state_witness_for_this_or_next_epoch` work
     /// since `get_epoch_chunk_producers` returns empty Vec which results

chain/client/src/stateless_validation/state_witness_producer.rs

@@ -1,4 +1,4 @@
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 
 use near_async::messaging::{CanSend, IntoSender};
@@ -337,33 +337,47 @@ impl Client {
         Ok(source_receipt_proofs)
     }
 
-    /// Sends the contract access to the same chunk validators that will receive the state witness for the chunk.
+    /// Sends the contract accesses to the same chunk validators
+    /// (except for the chunk producers that track the same shard),
+    /// which will receive the state witness for the new chunk.  
     fn send_contract_accesses_to_chunk_validators(
         &self,
         epoch_id: &EpochId,
         chunk_header: &ShardChunkHeader,
         contract_accesses: Vec<CodeHash>,
         my_signer: &ValidatorSigner,
     ) {
-        let chunk_validators = self
+        let chunk_production_key = ChunkProductionKey {
+            epoch_id: *epoch_id,
+            shard_id: chunk_header.shard_id(),
+            height_created: chunk_header.height_created(),
+        };
+
+        let chunk_validators: HashSet<AccountId> = self
             .epoch_manager
             .get_chunk_validator_assignments(
                 &epoch_id,
                 chunk_header.shard_id(),
                 chunk_header.height_created(),
             )
             .expect("Chunk validators must be defined")
-            .ordered_chunk_validators();
+            .assignments()
+            .iter()
+            .map(|(id, _)| id.clone())
+            .collect();
 
-        let chunk_production_key = ChunkProductionKey {
-            epoch_id: *epoch_id,
-            shard_id: chunk_header.shard_id(),
-            height_created: chunk_header.height_created(),
-        };
-        // TODO(#11099): Optimize the set of receivers by excluding self and chunk producers etc.
+        let chunk_producers: HashSet<AccountId> = self
+            .epoch_manager
+            .get_epoch_chunk_producers_for_shard(&epoch_id, chunk_header.shard_id())
+            .expect("Chunk producers must be defined")
+            .into_iter()
+            .collect();
+
+        let target_chunk_validators =
+            chunk_validators.difference(&chunk_producers).cloned().collect();
         self.network_adapter.send(PeerManagerMessageRequest::NetworkRequests(
             NetworkRequests::ChunkContractAccesses(
-                chunk_validators,
+                target_chunk_validators,
                 ChunkContractAccesses::new(chunk_production_key, contract_accesses, my_signer),
             ),
         ));

chain/epoch-manager/src/adapter.rs

@@ -215,6 +215,12 @@ pub trait EpochManagerAdapter: Send + Sync {
         epoch_id: &EpochId,
     ) -> Result<Vec<ValidatorStake>, EpochError>;
 
+    fn get_epoch_chunk_producers_for_shard(
+        &self,
+        epoch_id: &EpochId,
+        shard_id: ShardId,
+    ) -> Result<Vec<AccountId>, EpochError>;
+
     fn get_random_chunk_producer_for_shard(
         &self,
         epoch_id: &EpochId,
@@ -782,6 +788,15 @@ impl EpochManagerAdapter for EpochManagerHandle {
         Ok(epoch_manager.get_all_chunk_producers(epoch_id)?.to_vec())
     }
 
+    fn get_epoch_chunk_producers_for_shard(
+        &self,
+        epoch_id: &EpochId,
+        shard_id: ShardId,
+    ) -> Result<Vec<AccountId>, EpochError> {
+        let epoch_manager = self.read();
+        epoch_manager.get_epoch_chunk_producers_for_shard(epoch_id, shard_id)
+    }
+
     fn get_random_chunk_producer_for_shard(
         &self,
         epoch_id: &EpochId,

chain/epoch-manager/src/lib.rs

@@ -1074,6 +1074,27 @@ impl EpochManager {
         })
     }
 
+    /// Returns AccountIds of chunk producers that are assigned to a given shard-id in a given epoch.
+    pub fn get_epoch_chunk_producers_for_shard(
+        &self,
+        epoch_id: &EpochId,
+        shard_id: ShardId,
+    ) -> Result<Vec<AccountId>, EpochError> {
+        let epoch_info = self.get_epoch_info(&epoch_id)?;
+
+        let shard_layout = self.get_shard_layout(&epoch_id)?;
+        let shard_index = shard_layout.get_shard_index(shard_id);
+
+        let chunk_producers_settlement = epoch_info.chunk_producers_settlement();
+        let chunk_producers = chunk_producers_settlement
+            .get(shard_index)
+            .ok_or_else(|| EpochError::ShardingError(format!("invalid shard id {shard_id}")))?;
+        Ok(chunk_producers
+            .iter()
+            .map(|index| epoch_info.validator_account_id(*index).clone())
+            .collect())
+    }
+
     pub fn get_random_chunk_producer_for_shard(
         &self,
         epoch_id: &EpochId,