[Narwhal] Add checks to the DAG and panic if fail (MystenLabs#5173)

* Add checks to the DAG and panic if fail
* Take into account gc bound
* Add more log lines to understand
* Do not panic if the round does not exist

Co-authored-by: George Danezis <george@danez.is>

narwhal/consensus/src/bullshark.rs

@@ -7,8 +7,11 @@ use crate::{
 };
 use config::{Committee, Stake};
 use fastcrypto::{traits::EncodeDecodeBase64, Hash};
-use std::{collections::HashMap, sync::Arc};
-use tracing::debug;
+use std::{
+    collections::{BTreeSet, HashMap},
+    sync::Arc,
+};
+use tracing::{debug, error};
 use types::{Certificate, CertificateDigest, ConsensusStore, Round, SequenceNumber, StoreResult};
 
 #[cfg(test)]
@@ -35,6 +38,37 @@ impl ConsensusProtocol for Bullshark {
         let round = certificate.round();
         let mut consensus_index = consensus_index;
 
+        // We must have stored already the parents of this certiciate!
+        if round > 0 {
+            let parents = certificate.header.parents.clone();
+            if let Some(round_table) = state.dag.get(&(round - 1)) {
+                let store_parents: BTreeSet<&CertificateDigest> =
+                    round_table.iter().map(|(_, (digest, _))| digest).collect();
+
+                for parent_digest in parents {
+                    if !store_parents.contains(&parent_digest) {
+                        if round - 1 + self.gc_depth > state.last_committed_round {
+                            error!(
+                                "The store does not contain the parent of {:?}: Missing item digest={:?}",
+                                certificate, parent_digest
+                            );
+                        } else {
+                            debug!(
+                                "The store does not contain the parent of {:?}: Missing item digest={:?} (but below GC round)",
+                                certificate, parent_digest
+                            );
+                        }
+                    }
+                }
+            } else {
+                error!(
+                    "Round not present in Dag store: {:?} when looking for parents of {:?}",
+                    round - 1,
+                    certificate
+                );
+            }
+        }
+
         // Add the new certificate to the local storage.
         state
             .dag
@@ -90,6 +124,8 @@ impl ConsensusProtocol for Bullshark {
             .iter()
             .rev()
         {
+            debug!("Previous Leader {:?} has enough support", leader);
+
             // Starting from the oldest leader, flatten the sub-dag referenced by the leader.
             for x in utils::order_dag(self.gc_depth, leader, state) {
                 let digest = x.digest();

narwhal/consensus/src/consensus.rs

@@ -302,6 +302,8 @@ where
                     // Output the sequence in the right order.
                     for output in sequence {
                         let certificate = &output.certificate;
+                        tracing::debug!("Commit in Sequence {:?}", output);
+
                         #[cfg(not(feature = "benchmark"))]
                         if output.consensus_index % 5_000 == 0 {
                             tracing::debug!("Committed {}", certificate.header);