cleanup 🧹

Author: clabby

consensus/src/marshal/actor.rs

@@ -429,12 +429,13 @@ impl<
                             let block = self.get_finalized_block(height).await;
                             result.send(block).unwrap_or_else(|_| warn!(?height, "Failed to send block to orchestrator"));
                         }
-                        Orchestration::Processed { height, digest } => {
+                        Orchestration::Processed { height, digest, commitment } => {
                             // Update metrics
                             self.processed_height.set(height as i64);
 
                             // Cancel any outstanding requests (by height and by digest)
-                            resolver.cancel(Request::<B>::Block(digest)).await;
+                            resolver.cancel(Request::<B>::Block(commitment)).await;
+                            resolver.cancel(Request::<B>::CodingCommitment { height, digest }).await;
                             resolver.retain(Request::<B>::Finalized { height }.predicate()).await;
 
                             // If finalization exists, prune the archives
@@ -471,9 +472,9 @@ impl<
 
                             // Iterate backwards, repairing blocks as we go.
                             while cursor.height() > height {
-                                let commitment = cursor.parent();
-                                let Some(commitment) = shard_layer.get_digest(&commitment).await else {
-                                    dbg!("Missing block digest");
+                                let digest = cursor.parent();
+                                let Some(commitment) = shard_layer.get_commitment(&digest).await else {
+                                    resolver.fetch(Request::<B>::CodingCommitment { digest, height: cursor.height().saturating_sub(1) }).await;
                                     break;
                                 };
 
@@ -519,6 +520,14 @@ impl<
                                     };
                                     let _ = response.send(block.encode().into());
                                 }
+                                Request::CodingCommitment { digest, .. } => {
+                                    // Check for coding commitment locally
+                                    let Some(commitment) = shard_layer.get_commitment(&digest).await else {
+                                        debug!(?digest, "coding commitment missing on request");
+                                        continue;
+                                    };
+                                    let _ = response.send(commitment.encode().into());
+                                }
                                 Request::Finalized { height } => {
                                     // Get finalization
                                     let Some(finalization) = self.get_finalization_by_height(height).await else {
@@ -574,6 +583,24 @@ impl<
                                     debug!(?commitment, height, "received block");
                                     let _ = response.send(true);
                                 },
+                                Request::CodingCommitment { digest, height } => {
+                                    // Parse block digest and height
+                                    let Ok(commitment) = B::Commitment::decode_cfg(value.as_ref(), &()) else {
+                                        let _ = response.send(false);
+                                        continue;
+                                    };
+
+                                    // Persist the commitment.
+                                    shard_layer.put_commitment(height, digest, commitment).await;
+
+                                    // If we have the block, persist it and its finalization.
+                                    if let Some(block) = self.find_block(&mut shard_layer, commitment).await {
+                                        let finalization = self.cache.get_finalization_for(commitment).await;
+                                        self.finalize(block.height(), commitment, block.clone(), finalization, &mut notifier_tx).await;
+                                    }
+
+                                    let _ = response.send(true);
+                                },
                                 Request::Finalized { height } => {
                                     // Parse finalization
                                     let Ok((finalization, block)) = <(Finalization<V, B::Commitment>, B)>::decode_cfg(value, &((), self.codec_config.clone())) else {

consensus/src/marshal/envelope.rs

@@ -81,6 +81,7 @@ impl<B: Block, R: Spawner + Clock + Metrics + Storage, Z: Reporter<Activity = B>
                 // height is processed by the application), it is possible that the application may
                 // be asked to process a block it has already seen (which it can simply ignore).
                 let commitment = block.commitment();
+                let digest = block.digest();
                 self.application.report(block).await;
 
                 // Record that we have processed up through this height.
@@ -91,7 +92,9 @@ impl<B: Block, R: Spawner + Clock + Metrics + Storage, Z: Reporter<Activity = B>
                 }
 
                 // Notify the orchestrator that the block has been processed.
-                self.orchestrator.processed(height, commitment).await;
+                self.orchestrator
+                    .processed(height, digest, commitment)
+                    .await;
 
                 // Loop again without waiting for a notification (there may be more to process).
                 continue;