v4.0: blockstore: Add fallback in mark_slot_dead_if_not_full() (backport of #12026)

ledger/src/blockstore.rs

@@ -899,11 +899,14 @@ impl Blockstore {
         slot: Slot,
         shred_insertion_tracker: &mut ShredInsertionTracker,
     ) {
-        if shred_insertion_tracker
+        let mark_slot_dead = shred_insertion_tracker
             .slot_meta_working_set
             .get(&slot)
-            .is_none_or(|meta| !meta.new_slot_meta.borrow().is_full())
-        {
+            .map(|meta| !meta.new_slot_meta.borrow().is_full())
+            .or_else(|| self.meta(slot).ok().flatten().map(|meta| !meta.is_full()))
+            .unwrap_or(true);
+
+        if mark_slot_dead {
             // If the slot is already full there is no reason to mark as dead
             self.dead_slots_cf
                 .put_bytes_in_batch(
@@ -7713,6 +7716,43 @@ pub mod tests {
         assert!(blockstore.has_duplicate_shreds_in_slot(0));
     }
 
+    #[test]
+    fn test_mark_slot_dead_if_not_full() {
+        let ledger_path = get_tmp_ledger_path_auto_delete!();
+        let blockstore = Blockstore::open(ledger_path.path()).unwrap();
+
+        // Leave an empty slot
+        let empty_slot = 0;
+
+        // Insert a partial slot
+        let partial_slot = 5;
+        let (mut shreds, _) = make_slot_entries(partial_slot, partial_slot - 1, 100);
+        assert!(shreds.len() > 1);
+        shreds.pop();
+        blockstore.insert_shreds(shreds, None, false).unwrap();
+        assert!(!blockstore.meta(partial_slot).unwrap().unwrap().is_full());
+
+        // Insert a full slot
+        let full_slot = 10;
+        let (shreds, _) = make_slot_entries(full_slot, full_slot - 1, 100);
+        blockstore.insert_shreds(shreds, None, false).unwrap();
+        assert!(blockstore.meta(full_slot).unwrap().unwrap().is_full());
+
+        let mut shred_insertion_tracker =
+            ShredInsertionTracker::new(1, blockstore.db.batch().unwrap());
+
+        blockstore.mark_slot_dead_if_not_full(empty_slot, &mut shred_insertion_tracker);
+        blockstore.mark_slot_dead_if_not_full(partial_slot, &mut shred_insertion_tracker);
+        blockstore.mark_slot_dead_if_not_full(full_slot, &mut shred_insertion_tracker);
+        // Commit the write batch so state changes can be read back
+        blockstore
+            .write_batch(shred_insertion_tracker.write_batch)
+            .unwrap();
+        assert!(blockstore.is_dead(empty_slot));
+        assert!(blockstore.is_dead(partial_slot));
+        assert!(!blockstore.is_dead(full_slot));
+    }
+
     #[test]
     fn test_slot_data_iterator() {
         // Construct the shreds