v2.0: blockstore: mark slot as dead on data shred merkle root conflict (backport of #3970) (#4074)

* blockstore: mark slot as dead on data shred merkle root conflict (#3970)

(cherry picked from commit 5564a941ae65d6a43d8a4d97936fdc09792f92a8)

# Conflicts:
#	ledger/src/blockstore.rs

* fix conflicts

---------

Co-authored-by: Ashwin Sekar <ashwin@anza.xyz>
Co-authored-by: Ashwin Sekar <ashwin@solana.com>

ledger/src/blockstore.rs

@@ -1682,6 +1682,13 @@ impl Blockstore {
                     &shred,
                     duplicate_shreds,
                 ) {
+                    // This indicates there is an alternate version of this block.
+                    // Similar to the last index case above, we might never get all the
+                    // shreds for our current version, never replay this slot, and make no
+                    // progress. We cannot determine if we have the version that will eventually
+                    // be complete, so we take the conservative approach and mark the slot as dead
+                    // so that replay can dump and repair the correct version.
+                    write_batch.put::<cf::DeadSlots>(slot, &true).unwrap();
                     return Err(InsertDataShredError::InvalidShred);
                 }
             }
@@ -5848,21 +5855,6 @@ pub mod tests {
             .insert_shreds(shreds, None, false)
             .expect("Expected successful write of shreds");
 
-        let mut shreds1 = entries_to_test_shreds(
-            &entries[4..],
-            1,
-            0,
-            false,
-            0,
-            false, // merkle_variant
-        );
-        for (i, b) in shreds1.iter_mut().enumerate() {
-            b.set_index(8 + i as u32);
-        }
-        blockstore
-            .insert_shreds(shreds1, None, false)
-            .expect("Expected successful write of shreds");
-
         assert_eq!(
             blockstore.get_slot_entries(1, 0).unwrap()[2..4],
             entries[2..4],
@@ -7755,6 +7747,11 @@ pub mod tests {
             index
         );
 
+        // Block is now dead
+        blockstore.db.write(write_batch).unwrap();
+        assert!(blockstore.is_dead(slot));
+        blockstore.remove_dead_slot(slot).unwrap();
+
         // Blockstore should also have the merkle root meta of the original shred
         assert_eq!(
             blockstore
@@ -7786,6 +7783,7 @@ pub mod tests {
             fec_set_index + 30,
         );
 
+        let mut write_batch = blockstore.db.batch().unwrap();
         blockstore
             .check_insert_data_shred(
                 new_data_shred.clone(),
@@ -7802,23 +7800,25 @@ pub mod tests {
                 ShredSource::Turbine,
             )
             .unwrap();
+        blockstore.db.write(write_batch).unwrap();
 
         // Verify that we still have the merkle root meta for the original shred
         // and the new shred
-        assert_eq!(merkle_root_metas.len(), 2);
         assert_eq!(
-            merkle_root_metas
-                .get(&data_shred.erasure_set())
+            blockstore
+                .merkle_root_meta(data_shred.erasure_set())
                 .unwrap()
                 .as_ref()
+                .unwrap()
                 .merkle_root(),
             data_shred.merkle_root().ok()
         );
         assert_eq!(
-            merkle_root_metas
-                .get(&data_shred.erasure_set())
+            blockstore
+                .merkle_root_meta(data_shred.erasure_set())
                 .unwrap()
                 .as_ref()
+                .unwrap()
                 .first_received_shred_index(),
             index
         );