Fix marshal mailbox overflow coalescing

Author: clabby

consensus/src/marshal/core/mailbox.rs

@@ -187,8 +187,8 @@ impl<S: Scheme, V: Variant> Message<S, V> {
                 ..
             }
             | Self::GetFinalization { height, .. } => Some(*height) < current,
-            // Hints only inform the actor about heights strictly above the floor
-            Self::HintFinalized { height, .. } => Some(*height) <= current,
+            // Hints can fetch the current floor if the block is missing locally.
+            Self::HintFinalized { height, .. } => Some(*height) < current,
             // Durability acks cannot be dropped: callers depend on them
             Self::Proposed { .. } | Self::Verified { .. } | Self::Certified { .. } => false,
             // Digest and latest lookups are not bound to a specific height
@@ -270,7 +270,7 @@ impl<S: Scheme, V: Variant> Pending<S, V> {
 
     fn retain(&mut self) {
         let current = self.height();
-        self.hints.retain(|height, _| Some(*height) > current);
+        self.hints.retain(|height, _| Some(*height) >= current);
 
         let hints = &self.hints;
         self.messages.retain(|message| match message {
@@ -283,13 +283,23 @@ impl<S: Scheme, V: Variant> Pending<S, V> {
 
     fn set_floor(&mut self, height: Height, prune_archives: bool) {
         let current = self.height();
-        match &mut self.floor {
-            Some((pending, pending_prune)) if *pending > height => return,
-            Some((pending, pending_prune)) if *pending == height => {
-                *pending_prune |= prune_archives;
+        match self.floor.take() {
+            Some((pending, pending_prune)) if pending > height => {
+                self.floor = Some((pending, pending_prune));
+                return;
+            }
+            Some((pending, pending_prune)) if pending == height => {
+                self.floor = Some((height, pending_prune || prune_archives));
+            }
+            Some((pending, true)) if !prune_archives => {
+                self.prune = self.prune.max(Some(pending));
+                self.floor = Some((height, false));
+            }
+            Some(_) => {
+                self.floor = Some((height, prune_archives));
             }
-            floor => {
-                *floor = Some((height, prune_archives));
+            None => {
+                self.floor = Some((height, prune_archives));
             }
         }
 
@@ -352,9 +362,9 @@ impl<S: Scheme, V: Variant> Pending<S, V> {
     }
 
     fn hint_finalized(&mut self, height: Height, targets: Recipients<S::PublicKey>) {
-        // Hint is already covered by floor/prune
+        // Hints at the current floor may still need to fetch a missing block.
         let current = self.height();
-        if current.is_some_and(|current| height <= current) {
+        if current.is_some_and(|current| height < current) {
             return;
         }
 
@@ -890,9 +900,13 @@ mod tests {
     }
 
     fn set_floor(height: u64) -> TestMessage {
+        set_floor_prune(height, false)
+    }
+
+    fn set_floor_prune(height: u64, prune_archives: bool) -> TestMessage {
         TestMessage::SetFloor {
             height: Height::new(height),
-            prune_archives: false,
+            prune_archives,
         }
     }
 
@@ -1172,6 +1186,64 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn policy_retains_floor_height_hints() {
+        let mut overflow = pending();
+        let target = public_key(1);
+
+        <TestMessage as Policy>::handle(&mut overflow, set_floor(8));
+        <TestMessage as Policy>::handle(&mut overflow, hint_finalized(7, public_key(2)));
+        <TestMessage as Policy>::handle(&mut overflow, hint_finalized(8, target.clone()));
+        <TestMessage as Policy>::handle(&mut overflow, hint_finalized(9, public_key(3)));
+
+        assert!(hint_targets(&overflow, 7).is_none());
+        assert!(matches!(
+            hint_targets(&overflow, 8),
+            Some(Recipients::One(found)) if *found == target
+        ));
+        assert!(hint_targets(&overflow, 9).is_some());
+
+        let drained = drain(&mut overflow);
+        assert_eq!(drained.len(), 3);
+        assert!(matches!(
+            &drained[0],
+            TestMessage::SetFloor { height, .. } if *height == Height::new(8)
+        ));
+        assert!(matches!(
+            &drained[1],
+            TestMessage::HintFinalized { height, .. } if *height == Height::new(8)
+        ));
+        assert!(matches!(
+            &drained[2],
+            TestMessage::HintFinalized { height, .. } if *height == Height::new(9)
+        ));
+    }
+
+    #[test]
+    fn policy_preserves_floor_prune_when_higher_floor_supersedes_without_over_pruning() {
+        let mut overflow = pending();
+
+        <TestMessage as Policy>::handle(&mut overflow, set_floor_prune(10, true));
+        <TestMessage as Policy>::handle(&mut overflow, set_floor_prune(15, false));
+
+        assert_eq!(overflow.floor, Some((Height::new(15), false)));
+        assert_eq!(overflow.prune, Some(Height::new(10)));
+
+        let drained = drain(&mut overflow);
+        assert_eq!(drained.len(), 2);
+        assert!(matches!(
+            &drained[0],
+            TestMessage::SetFloor {
+                height,
+                prune_archives: false,
+            } if *height == Height::new(15)
+        ));
+        assert!(matches!(
+            &drained[1],
+            TestMessage::Prune { height } if *height == Height::new(10)
+        ));
+    }
+
     #[test]
     fn policy_replaces_floor_and_prune_and_drops_stale_pending_on_drain() {
         let mut overflow = pending();
@@ -1192,19 +1264,23 @@ mod tests {
             .push_back(PendingMessage::Message(get_block_8));
         <TestMessage as Policy>::handle(&mut overflow, set_floor(8));
         assert_eq!(overflow.floor, Some((Height::new(8), false)));
-        assert_eq!(overflow.messages.len(), 1);
+        assert_eq!(overflow.messages.len(), 2);
         assert!(!has_get_info(&overflow, 4));
         assert!(!has_get_block(&overflow, 7));
         assert!(has_get_block(&overflow, 8));
-        assert!(hint_targets(&overflow, 8).is_none());
+        assert!(hint_targets(&overflow, 8).is_some());
         let drained = drain(&mut overflow);
-        assert_eq!(drained.len(), 2);
+        assert_eq!(drained.len(), 3);
         assert!(matches!(
             &drained[0],
             TestMessage::SetFloor { height, .. } if *height == Height::new(8)
         ));
         assert!(matches!(
             &drained[1],
+            TestMessage::HintFinalized { height, .. } if *height == Height::new(8)
+        ));
+        assert!(matches!(
+            &drained[2],
             TestMessage::GetBlock {
                 identifier: Identifier::Height(height),
                 ..