Feat(near-client): Chunk distribution via message bus

chain/chunks/src/adapter.rs

@@ -8,7 +8,7 @@ use near_primitives::{
     types::EpochId,
 };
 
-#[derive(Message, Debug, strum::IntoStaticStr)]
+#[derive(Message, Debug, strum::IntoStaticStr, PartialEq)]
 #[rtype(result = "()")]
 pub enum ShardsManagerRequestFromClient {
     /// Processes the header seen from a block we received, if we have not already received the
@@ -50,4 +50,20 @@ pub enum ShardsManagerRequestFromClient {
     /// proofs, but cannot be marked as complete because the previous block isn't available),
     /// and completes them if so.
     CheckIncompleteChunks(CryptoHash),
+    /// Process a `PartialEncodedChunk` obtained via the Chunk Distribution Network feature.
+    /// If the chunk turns out to be invalid then request from the p2p network instead.
+    ProcessOrRequestChunk {
+        candidate_chunk: PartialEncodedChunk,
+        request_header: ShardChunkHeader,
+        prev_hash: CryptoHash,
+    },
+    /// Similar to `ProcessOrRequestChunk` but for orphan chunks.
+    /// Process a `PartialEncodedChunk` obtained via the Chunk Distribution Network feature.
+    /// If the chunk turns out to be invalid then request from the p2p network instead.
+    ProcessOrRequestChunkForOrphan {
+        candidate_chunk: PartialEncodedChunk,
+        request_header: ShardChunkHeader,
+        epoch_id: EpochId,
+        ancestor_hash: CryptoHash,
+    },
 }

chain/chunks/src/lib.rs

@@ -2062,6 +2062,27 @@ impl ShardsManager {
             ShardsManagerRequestFromClient::CheckIncompleteChunks(prev_block_hash) => {
                 self.check_incomplete_chunks(&prev_block_hash)
             }
+            ShardsManagerRequestFromClient::ProcessOrRequestChunk {
+                candidate_chunk,
+                request_header,
+                prev_hash,
+            } => {
+                if let Err(err) = self.process_partial_encoded_chunk(candidate_chunk.into()) {
+                    warn!(target: "chunks", ?err, "Error processing partial encoded chunk");
+                    self.request_chunk_single(&request_header, prev_hash, false);
+                }
+            }
+            ShardsManagerRequestFromClient::ProcessOrRequestChunkForOrphan {
+                candidate_chunk,
+                request_header,
+                ancestor_hash,
+                epoch_id,
+            } => {
+                if let Err(e) = self.process_partial_encoded_chunk(candidate_chunk.into()) {
+                    warn!(target: "chunks", "Error processing partial encoded chunk: {:?}", e);
+                    self.request_chunks_for_orphan(vec![request_header], &epoch_id, ancestor_hash);
+                }
+            }
         }
     }