Add test to ensure data shreds with empty data would be inserted (#16955)

* Add test to ensure data shreds with empty data would be inserted

* Add error log statements for malformed shreds

ledger/src/blockstore.rs

@@ -1292,9 +1292,37 @@ impl Blockstore {
         };
 
         if shred.data_header.size == 0 {
+            let leader_pubkey = leader_schedule
+                .and_then(|leader_schedule| leader_schedule.slot_leader_at(slot, None));
+
+            datapoint_error!(
+                "blockstore_error",
+                (
+                    "error",
+                    format!(
+                        "Leader {:?}, slot {}: received index {} is empty",
+                        leader_pubkey, slot, shred_index,
+                    ),
+                    String
+                )
+            );
             return false;
         }
         if shred.payload.len() > SHRED_PAYLOAD_SIZE {
+            let leader_pubkey = leader_schedule
+                .and_then(|leader_schedule| leader_schedule.slot_leader_at(slot, None));
+
+            datapoint_error!(
+                "blockstore_error",
+                (
+                    "error",
+                    format!(
+                        "Leader {:?}, slot {}: received index {} shred.payload.len() > SHRED_PAYLOAD_SIZE",
+                        leader_pubkey, slot, shred_index,
+                    ),
+                    String
+                )
+            );
             return false;
         }
 
@@ -1303,8 +1331,7 @@ impl Blockstore {
         let last_index = slot_meta.last_index;
         if shred_index >= last_index {
             let leader_pubkey = leader_schedule
-                .map(|leader_schedule| leader_schedule.slot_leader_at(slot, None))
+                .and_then(|leader_schedule| leader_schedule.slot_leader_at(slot, None));
-                .unwrap_or(None);
 
             let ending_shred: Cow<Vec<u8>> = self.get_data_shred_from_just_inserted_or_db(
                 just_inserted_data_shreds,
@@ -1340,8 +1367,7 @@ impl Blockstore {
         // less than our current received
         if last_in_slot && shred_index < slot_meta.received {
             let leader_pubkey = leader_schedule
-                .map(|leader_schedule| leader_schedule.slot_leader_at(slot, None))
+                .and_then(|leader_schedule| leader_schedule.slot_leader_at(slot, None));
-                .unwrap_or(None);
 
             let ending_shred: Cow<Vec<u8>> = self.get_data_shred_from_just_inserted_or_db(
                 just_inserted_data_shreds,
@@ -5303,17 +5329,46 @@ pub mod tests {
             let mut shred5 = shreds[5].clone();
             shred5.payload.push(10);
             shred5.data_header.size = shred5.payload.len() as u16;
-            assert_eq!(
+            assert!(!blockstore.should_insert_data_shred(
-                blockstore.should_insert_data_shred(
                 &shred5,
                 &slot_meta,
                 &HashMap::new(),
                 &last_root,
                 None,
                 false
-                ),
+            ));
-                false
+
+            // Ensure that an empty shred (one with no data) would get inserted. Such shreds
+            // may be used as signals (broadcast does so to indicate a slot was interrupted)
+            // Reuse shred5's header values to avoid a false negative result
+            let mut empty_shred = Shred::new_from_data(
+                shred5.common_header.slot,
+                shred5.common_header.index,
+                shred5.data_header.parent_offset,
+                None, // data
+                true, // is_last_data
+                true, // is_last_in_slot
+                0,    // reference_tick
+                shred5.common_header.version,
+                shred5.common_header.fec_set_index,
             );
+            assert!(blockstore.should_insert_data_shred(
+                &empty_shred,
+                &slot_meta,
+                &HashMap::new(),
+                &last_root,
+                None,
+                false
+            ));
+            empty_shred.data_header.size = 0;
+            assert!(!blockstore.should_insert_data_shred(
+                &empty_shred,
+                &slot_meta,
+                &HashMap::new(),
+                &last_root,
+                None,
+                false
+            ));
 
             // Trying to insert another "is_last" shred with index < the received index should fail
             // skip over shred 7