fast leader handover: error on bad UpdateParent during shred ingest (#605)

#### Problem and Summary of Changes
When disseminating an `UpdateParent` component, we need to ensure that
the parent change passes validity checks with respect to the
`BlockHeader`.

We explicitly cover the four cases:
- `UpdateParent` observed before `BlockHeader` and...
- ... `UpdateParent` and `BlockHeader` are in the same shred insertion
batch
- ... `UpdateParent` and `BlockHeader` are in different shred insertion
batches

- `BlockHeader` observed before `UpdateParent` and...
- ... `UpdateParent` and `BlockHeader` are in the same shred insertion
batch
- ... `UpdateParent` and `BlockHeader` are in different shred insertion
batches

These checks are efficient, since we can detect the presence of a
`BlockHeader` in a shred's payload without having to touch blockstore.

Author: ksn6

ledger/src/blockstore.rs

@@ -1954,47 +1954,44 @@ impl Blockstore {
         None
     }
 
-    fn maybe_insert_parent_meta(
-        &self,
-        current_shred: &Shred,
-        entry: &mut ParentMetaWorkingSetEntry,
-    ) {
+    fn maybe_parse_block_header(&self, current_shred: &Shred) -> Option<ParentMeta> {
+        // Block headers only occur at index 0
+        if current_shred.index() != 0 {
+            return None;
+        }
+
         // The contents of BlockHeaderV1 entirely fit into a single shred. So, let's just parse the
         // first shred.
         let shred_bytes = current_shred.payload();
         let Ok(payload) = shred::wire::get_data(shred_bytes) else {
-            return;
+            return None;
         };
 
         // Check whether this is a BlockMarker
         if !BlockComponent::infer_is_block_marker(payload).unwrap_or(false) {
-            return;
+            return None;
         }
 
         // Try to parse BlockHeader from the payload
-        let Some((parent_slot, parent_block_id)) =
-            BlockComponent::parse_block_header_from_data_payload(payload)
-        else {
-            return;
-        };
+        let (parent_slot, parent_block_id) =
+            BlockComponent::parse_block_header_from_data_payload(payload)?;
 
         let parent_meta = ParentMeta {
             parent_slot,
             parent_block_id,
             replay_fec_set_index: 0,
         };
 
-        entry.parent_meta = Some(WorkingEntry::Dirty(parent_meta));
+        Some(parent_meta)
     }
 
-    fn maybe_insert_update_parent(
+    fn maybe_parse_update_parent(
         &self,
         current_shred: &Shred,
         slot: Slot,
         location: BlockLocation,
         just_inserted_shreds: &HashMap<(BlockLocation, ShredId), Cow<'_, Shred>>,
-        entry: &mut ParentMetaWorkingSetEntry,
-    ) -> bool {
+    ) -> Option<ParentMeta> {
         let current_index = current_shred.index();
         let fec_set_index = current_shred.fec_set_index();
         let data_complete = current_shred.data_complete();
@@ -2032,41 +2029,30 @@ impl Blockstore {
 
             (shred_bytes, fec_set_index)
         } else {
-            return false;
+            return None;
         };
 
         // Process the shred bytes if we have them
-        let Some(shred_bytes) = shred_bytes else {
-            return false;
-        };
-
-        let Ok(payload) = shred::wire::get_data(&shred_bytes) else {
-            return false;
-        };
+        let shred_bytes = shred_bytes?;
+        let payload = shred::wire::get_data(&shred_bytes).ok()?;
 
         // Check whether this is a BlockMarker
         if !BlockComponent::infer_is_block_marker(payload).unwrap_or(false) {
-            return false;
+            return None;
         }
 
         // Try to parse UpdateParent from the payload
-        let Some((new_parent_slot, new_parent_block_id)) =
-            BlockComponent::parse_update_parent_from_data_payload(payload)
-        else {
-            return false;
-        };
+        let (new_parent_slot, new_parent_block_id) =
+            BlockComponent::parse_update_parent_from_data_payload(payload)?;
 
         // First time seeing this UpdateParent
-        let parent_meta = ParentMeta {
+        let update_parent_meta = ParentMeta {
             parent_slot: new_parent_slot,
             parent_block_id: new_parent_block_id,
             replay_fec_set_index: target_fec_set_index,
         };
 
-        // Store the UpdateParent metadata
-        entry.parent_meta = Some(WorkingEntry::Dirty(parent_meta));
-
-        true
+        Some(update_parent_meta)
     }
 
     /// Create an entry to the specified `write_batch` that performs shred
@@ -2250,6 +2236,75 @@ impl Blockstore {
         Ok(())
     }
 
+    /// Validates compatibility between two ParentMetas.
+    ///
+    /// This function determines which ParentMeta is which (UpdateParent vs BlockHeader) and
+    /// validates that they are compatible. Works irrespective of parameter order.
+    ///
+    /// Returns:
+    /// - `Ok(true)` if validation passed and the new ParentMeta should be written
+    /// - `Ok(false)` if the new ParentMeta should NOT be written (same type and match, or UpdateParent takes precedence)
+    /// - `Err` if validation fails
+    fn should_write_parent_meta(slot: Slot, new: &ParentMeta, prev: &ParentMeta) -> Result<bool> {
+        // Determine which is the UpdateParent and which is the BlockHeader
+        let (update_parent_meta, block_header_meta, should_write) = match (
+            new.populated_from_block_header(),
+            prev.populated_from_block_header(),
+        ) {
+            // New is BlockHeader, prev is UpdateParent
+            // UpdateParent takes precedence, so don't overwrite
+            (true, false) => (prev, new, false),
+            // New is UpdateParent, prev is BlockHeader
+            // Validate and allow UpdateParent to replace BlockHeader
+            (false, true) => (new, prev, true),
+            // Both are the same type - ensure they match
+            _ => match new == prev {
+                true => return Ok(false),
+                false => return Err(BlockstoreError::BlockComponentMismatch(slot)),
+            },
+        };
+
+        // Validate that the UpdateParent is compatible with the BlockHeader
+        if update_parent_meta.block() == block_header_meta.block() {
+            return Err(BlockstoreError::UpdateParentMatchesBlockHeader(slot));
+        }
+
+        if update_parent_meta.parent_slot > block_header_meta.parent_slot {
+            return Err(BlockstoreError::UpdateParentSlotGreaterThanBlockHeader(
+                slot,
+            ));
+        }
+
+        Ok(should_write)
+    }
+
+    /// Fetches or retrieves the existing ParentMeta for a given slot and location.
+    ///
+    /// This method first checks the working set, and if not found there, fetches from blockstore.
+    /// Returns the existing ParentMeta if it exists, or None otherwise.
+    fn get_or_fetch_parent_meta(
+        &self,
+        slot: Slot,
+        location: BlockLocation,
+        entry: &mut ParentMetaWorkingSetEntry,
+    ) -> Result<Option<ParentMeta>> {
+        // First check working set
+        if let Some(parent_meta) = entry.parent_meta.as_ref() {
+            return Ok(Some(*parent_meta.as_ref()));
+        }
+
+        // Then fetch from blockstore if needed
+        if !entry.fetched_from_blockstore {
+            entry.fetched_from_blockstore = true;
+            if let Some(parent_meta) = self.parent_meta_cf.get((slot, location))? {
+                entry.parent_meta = Some(WorkingEntry::Clean(parent_meta));
+                return Ok(Some(parent_meta));
+            }
+        }
+
+        Ok(None)
+    }
+
     fn maybe_update_parent_meta(
         &self,
         shred: &Shred,
@@ -2261,42 +2316,41 @@ impl Blockstore {
         let key = (location, slot);
         let entry = parent_meta_working_set.entry(key).or_default();
 
-        // If the working set ParentMeta exists and has an UpdateParent, then return early
-        if let Some(parent_meta) = entry.parent_meta.as_ref() {
-            let populated_from_update_parent = match parent_meta {
-                WorkingEntry::Dirty(parent_meta) | WorkingEntry::Clean(parent_meta) => {
-                    parent_meta.populated_from_update_parent()
-                }
-            };
-
-            if populated_from_update_parent {
-                return Ok(());
+        // Fetch or retrieve existing ParentMeta (from working set or blockstore)
+        let previous_parent_meta = self.get_or_fetch_parent_meta(slot, location, entry)?;
+
+        // Try to parse new ParentMeta from the current shred
+        // - If shred.index() == 0, try to parse as a block header
+        // - Otherwise, try to parse as an UpdateParent (only if we don't already have one,
+        //   since UpdateParent takes precedence and parsing is expensive)
+        let new_parent_meta = match (shred.index(), &previous_parent_meta) {
+            // Always try to parse block header at index 0
+            (0, _) => self.maybe_parse_block_header(shred),
+            // Try to parse UpdateParent only if we don't have one already
+            (_, Some(pm)) if pm.populated_from_block_header() => {
+                self.maybe_parse_update_parent(shred, slot, location, just_inserted_shreds)
             }
-        }
-        // If we haven't hit blockstore for this key yet...
-        else if !entry.fetched_from_blockstore {
-            entry.fetched_from_blockstore = true;
-
-            if let Some(parent_meta) = self.parent_meta_cf.get((slot, location))? {
-                entry.parent_meta = Some(WorkingEntry::Clean(parent_meta));
-
-                if parent_meta.populated_from_update_parent() {
-                    return Ok(());
-                }
+            (_, None) => {
+                self.maybe_parse_update_parent(shred, slot, location, just_inserted_shreds)
             }
-        }
+            // previous_parent_meta is UpdateParent; don't parse another one
+            _ => None,
+        };
 
-        // Run through insertion logic for update_parent. If we succeed at inserting an
-        // UpdateParent, then return.
-        if self.maybe_insert_update_parent(shred, slot, location, just_inserted_shreds, entry) {
+        let Some(new_parent_meta) = new_parent_meta else {
             return Ok(());
-        }
+        };
 
-        // If there isn't any parent_meta, then attempt to detect a block header.
-        if entry.parent_meta.is_none() && shred.index() == 0 {
-            self.maybe_insert_parent_meta(shred, entry);
+        // Validate new ParentMeta against previous (if both exist)
+        if let Some(prev) = &previous_parent_meta {
+            if !Self::should_write_parent_meta(slot, &new_parent_meta, prev)? {
+                return Ok(());
+            }
         }
 
+        // Update entry with new ParentMeta
+        entry.parent_meta = Some(WorkingEntry::Dirty(new_parent_meta));
+
         Ok(())
     }
 

ledger/src/blockstore/error.rs

@@ -58,5 +58,13 @@ pub enum BlockstoreError {
     MissingMerkleRoot(Slot, u64),
     #[error("BlockComponent misalignment slot {0}, index {1}")]
     BlockComponentMisalignment(Slot, u64),
+    #[error("Update parent matches block header slot {0}")]
+    UpdateParentMatchesBlockHeader(Slot),
+    #[error("Update parent slot greater than block header slot {0}")]
+    UpdateParentSlotGreaterThanBlockHeader(Slot),
+    #[error("Unexpected block component")]
+    UnexpectedBlockComponent,
+    #[error("Block component mismatch slot {0}")]
+    BlockComponentMismatch(Slot),
 }
 pub type Result<T> = std::result::Result<T, BlockstoreError>;

ledger/src/blockstore_meta.rs

@@ -990,6 +990,14 @@ impl ParentMeta {
     pub fn populated_from_update_parent(&self) -> bool {
         self.replay_fec_set_index > 0
     }
+
+    pub fn populated_from_block_header(&self) -> bool {
+        self.replay_fec_set_index == 0
+    }
+
+    pub fn block(&self) -> (Slot, Hash) {
+        (self.parent_slot, self.parent_block_id)
+    }
 }
 
 #[cfg(test)]