Fix SlotMeta is_connected tracking

ledger/src/blockstore.rs

@@ -3315,6 +3315,54 @@ impl Blockstore {
         );
         Ok(())
     }
+
+    /// Mark a root `slot` as connected, traverse `slot`'s children and update
+    /// the children's connected status if appropriate.
+    ///
+    /// A ledger with a full path of blocks from genesis to the latest root will
+    /// have all of the rooted blocks marked as connected such that new blocks
+    /// could also be connected. However, starting from some root (such as from
+    /// a snapshot) is a valid way to join a cluster. For this case, mark this
+    /// root as connected such that the node that joined midway through can
+    /// have their slots considered connected.
+    pub fn set_and_chain_connected_on_root_and_next_slots(&self, root: Slot) -> Result<()> {
+        let mut root_meta = self
+            .meta(root)?
+            .unwrap_or_else(|| SlotMeta::new(root, None));
+        // If the slot was already connected, there is nothing to do as this slot's
+        // children are also assumed to be appropriately connected
+        let already_connected = root_meta.is_connected();
+        if already_connected {
+            return Ok(());
+        }
+        info!(
+            "Marking slot {} and any full children slots as connected",
+            root
+        );
+
+        // Mark both bits as connected for snapshots slots so that these slots
+        // behavior more similarly to typical slots that become connected.
+        root_meta.set_parent_connected();
+        root_meta.set_connected();
+        let mut write_batch = self.db.batch()?;
+        write_batch.put::<cf::SlotMeta>(root_meta.slot, &root_meta)?;
+
+        let mut next_slots = VecDeque::from(root_meta.next_slots);
+        while !next_slots.is_empty() {
+            let slot = next_slots.pop_front().unwrap();
+            let mut meta = self.meta(slot)?.unwrap_or_else(|| {
+                panic!("Slot {} is a child but has no SlotMeta in blockstore", slot)
+            });
+
+            if update_child_from_connected_parent(&mut meta) {
+                next_slots.extend(meta.next_slots.iter());
+            }
+            write_batch.put::<cf::SlotMeta>(meta.slot, &meta)?;
+        }
+
+        self.db.write(write_batch)?;
+        Ok(())
+    }
 }
 
 // Update the `completed_data_indexes` with a new shred `new_shred_index`. If a
@@ -3729,32 +3777,22 @@ fn handle_chaining_for_slot(
         }
     }
 
-    // If this is a newly inserted slot, then we know the children of this slot were not previously
+    // If this is a newly completed slot, then we know the children of this slot were not previously
     // connected to the trunk of the ledger. Thus if slot.is_connected is now true, we need to
     // update all child slots with `is_connected` = true because these children are also now newly
     // connected to trunk of the ledger
     let should_propagate_is_connected =
         is_newly_completed_slot(&RefCell::borrow(meta), meta_backup)
-            && RefCell::borrow(meta).is_connected;
+            && RefCell::borrow(meta).is_parent_connected();
 
     if should_propagate_is_connected {
-        // slot_function returns a boolean indicating whether to explore the children
-        // of the input slot
-        let slot_function = |slot: &mut SlotMeta| {
-            slot.is_connected = true;
-
-            // We don't want to set the is_connected flag on the children of non-full
-            // slots
-            slot.is_full()
-        };
-
         traverse_children_mut(
             db,
             slot,
             meta,
             working_set,
             new_chained_slots,
-            slot_function,
+            update_child_from_connected_parent,
         )?;
     }
 
@@ -3808,21 +3846,44 @@ where
     Ok(())
 }
 
+// Update the meta for a slot whose parent is known to be connected.
+// Returns a bool indicating whether this slot is now connected which would
+// indicate that it's children are also in need of update.
+fn update_child_from_connected_parent(meta: &mut SlotMeta) -> bool {
+    // If this slot is already connected, its' children should
+    // also already be connected so cut the traversal here.
+    if meta.is_connected() {
+        return false;
+    }
+    meta.set_parent_connected();
+
+    // This slot's parent is known to be connected, so if
+    // it is also full, then the slot is now connected.
+    let is_full = meta.is_full();
+    if is_full {
+        meta.set_connected();
+    }
+
+    // Traverse children if we are full
+    is_full
+}
+
 fn is_orphan(meta: &SlotMeta) -> bool {
     // If we have no parent, then this is the head of a detached chain of
     // slots
     meta.parent_slot.is_none()
 }
 
 // 1) Chain current_slot to the previous slot defined by prev_slot_meta
-// 2) Determine whether to set the is_connected flag
 fn chain_new_slot_to_prev_slot(
     prev_slot_meta: &mut SlotMeta,
     current_slot: Slot,
     current_slot_meta: &mut SlotMeta,
 ) {
     prev_slot_meta.next_slots.push(current_slot);
-    current_slot_meta.is_connected = prev_slot_meta.is_connected && prev_slot_meta.is_full();
+    if prev_slot_meta.is_connected() {
+        current_slot_meta.set_parent_connected();
+    }
 }
 
 fn is_newly_completed_slot(slot_meta: &SlotMeta, backup_slot_meta: &Option<SlotMeta>) -> bool {
@@ -3835,8 +3896,8 @@ fn slot_has_updates(slot_meta: &SlotMeta, slot_meta_backup: &Option<SlotMeta>) -
     // We should signal that there are updates if we extended the chain of consecutive blocks starting
     // from block 0, which is true iff:
     // 1) The block with index prev_block_index is itself part of the trunk of consecutive blocks
-    // starting from block 0,
-    slot_meta.is_connected &&
+    // starting from block 0 or from a snapshot
+    slot_meta.is_parent_connected() &&
         // AND either:
         // 1) The slot didn't exist in the database before, and now we have a consecutive
         // block for that slot
@@ -4822,7 +4883,7 @@ pub mod tests {
         assert_eq!(meta.parent_slot, Some(0));
         assert_eq!(meta.last_index, Some(num_shreds - 1));
         assert!(meta.next_slots.is_empty());
-        assert!(meta.is_connected);
+        assert!(meta.is_connected());
     }
 
     #[test]
@@ -5345,7 +5406,7 @@ pub mod tests {
         let meta1 = blockstore.meta(1).unwrap().unwrap();
         assert!(meta1.next_slots.is_empty());
         // Slot 1 is not trunk because slot 0 hasn't been inserted yet
-        assert!(!meta1.is_connected);
+        assert!(!meta1.is_connected());
         assert_eq!(meta1.parent_slot, Some(0));
         assert_eq!(meta1.last_index, Some(shreds_per_slot as u64 - 1));
 
@@ -5357,15 +5418,15 @@ pub mod tests {
         let meta2 = blockstore.meta(2).unwrap().unwrap();
         assert!(meta2.next_slots.is_empty());
         // Slot 2 is not trunk because slot 0 hasn't been inserted yet
-        assert!(!meta2.is_connected);
+        assert!(!meta2.is_connected());
         assert_eq!(meta2.parent_slot, Some(1));
         assert_eq!(meta2.last_index, Some(shreds_per_slot as u64 - 1));
 
         // Check the first slot again, it should chain to the second slot,
         // but still isn't part of the trunk
         let meta1 = blockstore.meta(1).unwrap().unwrap();
         assert_eq!(meta1.next_slots, vec![2]);
-        assert!(!meta1.is_connected);
+        assert!(!meta1.is_connected());
         assert_eq!(meta1.parent_slot, Some(0));
         assert_eq!(meta1.last_index, Some(shreds_per_slot as u64 - 1));
 
@@ -5384,12 +5445,13 @@ pub mod tests {
                 assert_eq!(meta.parent_slot, Some(slot - 1));
             }
             assert_eq!(meta.last_index, Some(shreds_per_slot as u64 - 1));
-            assert!(meta.is_connected);
+            assert!(meta.is_connected());
         }
     }
 
     #[test]
     fn test_handle_chaining_missing_slots() {
+        solana_logger::setup();
         let ledger_path = get_tmp_ledger_path_auto_delete!();
         let blockstore = Blockstore::open(ledger_path.path()).unwrap();
 
@@ -5442,10 +5504,11 @@ pub mod tests {
                 assert_eq!(meta.parent_slot, Some(slot - 1));
             }
 
+            assert!(!meta.is_connected());
             if slot == 0 {
-                assert!(meta.is_connected);
+                assert!(meta.is_parent_connected());
             } else {
-                assert!(!meta.is_connected);
+                assert!(!meta.is_parent_connected());
             }
         }
 
@@ -5470,13 +5533,14 @@ pub mod tests {
                 assert_eq!(meta.parent_slot, Some(slot - 1));
             }
             assert_eq!(meta.last_index, Some(shreds_per_slot as u64 - 1));
-            assert!(meta.is_connected);
+            assert!(meta.is_connected());
         }
     }
 
     #[test]
     #[allow(clippy::cognitive_complexity)]
     pub fn test_forward_chaining_is_connected() {
+        solana_logger::setup();
         let ledger_path = get_tmp_ledger_path_auto_delete!();
         let blockstore = Blockstore::open(ledger_path.path()).unwrap();
 
@@ -5513,14 +5577,14 @@ pub mod tests {
             }
 
             // Ensure that each slot has their parent correct
-            // Additionally, slot 0 should be the only connected slot
             if slot == 0 {
                 assert_eq!(meta.parent_slot, Some(0));
-                assert!(meta.is_connected);
             } else {
                 assert_eq!(meta.parent_slot, Some(slot - 1));
-                assert!(!meta.is_connected);
             }
+            // No slots should be connected yet, not even slot 0
+            // as slot 0 is still not full yet
+            assert!(!meta.is_connected());
 
             assert_eq!(meta.last_index, Some(shreds_per_slot as u64 - 1));
         }
@@ -5539,10 +5603,11 @@ pub mod tests {
                     } else {
                         assert!(meta.next_slots.is_empty());
                     }
-                    if slot <= slot_index as u64 + 3 {
-                        assert!(meta.is_connected);
+                    if slot < slot_index as u64 + 3 {
+                        assert!(meta.is_full());
+                        assert!(meta.is_connected());
                     } else {
-                        assert!(!meta.is_connected);
+                        assert!(!meta.is_connected());
                     }
 
                     if slot == 0 {
@@ -5556,6 +5621,67 @@ pub mod tests {
             }
         }
     }
+
+    #[test]
+    fn test_set_and_chain_connected_on_root_and_next_slots() {
+        let ledger_path = get_tmp_ledger_path_auto_delete!();
+        let blockstore = Blockstore::open(ledger_path.path()).unwrap();
+
+        // Create enough entries to ensure 5 shreds result
+        let entries_per_slot = max_ticks_per_n_shreds(5, None);
+
+        // Start a chain from a slot not in blockstore, such is the case when
+        // a node starts with no blockstore and downloads a snapshot. This is
+        // the special case and slot will be marked connected despite its'
+        // parent not being connected (or existing) and not being full
+        blockstore
+            .set_and_chain_connected_on_root_and_next_slots(5)
+            .unwrap();
+        let slot_meta5 = blockstore.meta(5).unwrap().unwrap();
+        assert!(!slot_meta5.is_full());
+        assert!(slot_meta5.is_parent_connected());
+        assert!(slot_meta5.is_connected());
+
+        // Insert some slots and ensure they connect from slot 5 correctly
+        let (shreds, _) = make_many_slot_entries(6, 5, entries_per_slot);
+        blockstore.insert_shreds(shreds, None, false).unwrap();
+        for slot in 6..=10 {
+            let meta = blockstore.meta(slot).unwrap().unwrap();
+            assert!(meta.is_parent_connected());
+            assert!(meta.is_connected());
+        }
+
+        // Chain connected on slots that are already connected, should just noop
+        blockstore
+            .set_and_chain_connected_on_root_and_next_slots(6)
+            .unwrap();
+        for slot in 6..=10 {
+            let meta = blockstore.meta(slot).unwrap().unwrap();
+            assert!(meta.is_parent_connected());
+            assert!(meta.is_connected());
+        }
+
+        // Insert some slots that are not connected and then ensure they connect
+        // This might be the case for a node that had existing blockstore data
+        // prior to the is_connected behavior being expanded to snapshot slots
+        let (shreds, _) = make_many_slot_entries(15, 5, entries_per_slot);
+        blockstore.insert_shreds(shreds, None, false).unwrap();
+        for slot in 15..=19 {
+            let meta = blockstore.meta(slot).unwrap().unwrap();
+            assert!(!meta.is_parent_connected());
+            assert!(!meta.is_connected());
+        }
+        blockstore
+            .set_and_chain_connected_on_root_and_next_slots(15)
+            .unwrap();
+        for slot in 15..=19 {
+            let meta = blockstore.meta(slot).unwrap().unwrap();
+            // Slot 15 doesn't have a parent so parent is not connected
+            assert!(meta.is_parent_connected() || slot == 15);
+            assert!(meta.is_connected());
+        }
+    }
+
     /*
         #[test]
         pub fn test_chaining_tree() {
@@ -5622,7 +5748,7 @@ pub mod tests {
                 let slot_meta = blockstore.meta(slot).unwrap().unwrap();
                 assert_eq!(slot_meta.consumed, entries_per_slot);
                 assert_eq!(slot_meta.received, entries_per_slot);
-                assert!(slot_meta.is_connected);
+                assert!(slot_meta.is_connected;
                 let slot_parent = {
                     if slot == 0 {
                         0