adress code review

Author: numinnex

core/binary_protocol/src/consensus/header.rs

@@ -70,6 +70,14 @@ pub fn read_size_field(header: &[u8]) -> Option<u32> {
 pub trait ConsensusHeader: Sized + CheckedBitPattern + NoUninit {
     const COMMAND: Command2;
 
+    /// Whether a frame carrying `command` may be typed as this header.
+    /// Defaults to an exact match; a header that serves several commands
+    /// with one layout (e.g. `RepairDone` / `RangeEvicted`) widens it.
+    #[must_use]
+    fn accepts(command: Command2) -> bool {
+        command == Self::COMMAND
+    }
+
     /// # Errors
     /// Returns `ConsensusError` if the header fields are inconsistent.
     fn validate(&self) -> Result<(), ConsensusError>;
@@ -1215,6 +1223,12 @@ const _: () = {
 
 impl ConsensusHeader for RepairRangeReplyHeader {
     const COMMAND: Command2 = Command2::RepairDone;
+    // One layout, two commands: `RepairDone` terminates a stream,
+    // `RangeEvicted` prefixes it. Without this widening, `try_into_typed`
+    // rejects `RangeEvicted` frames before `validate` ever sees them.
+    fn accepts(command: Command2) -> bool {
+        command == Command2::RepairDone || command == Command2::RangeEvicted
+    }
     fn operation(&self) -> Operation {
         Operation::Reserved
     }

core/consensus/src/impls.rs

@@ -580,7 +580,9 @@ pub enum VsrAction {
     },
     /// Broadcast a `RequestStartView` probe (recovering replica asking for
     /// the current view's `StartView`; only that view's primary answers).
-    SendRequestStartView { namespace: u64 },
+    /// Stamped with the prober's view so peers can fence stale duplicates
+    /// out of the probed-primary election path.
+    SendRequestStartView { view: u32, namespace: u64 },
     /// Send `StartView` to all backups (as new primary).
     SendStartView {
         view: u32,
@@ -1236,6 +1238,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
                             .borrow_mut()
                             .reset(TimeoutKind::RequestStartViewMessage);
                         actions.push(VsrAction::SendRequestStartView {
+                            view: self.view.get(),
                             namespace: self.namespace,
                         });
                     }
@@ -1245,6 +1248,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
                         .borrow_mut()
                         .reset(TimeoutKind::RequestStartViewMessage);
                     actions.push(VsrAction::SendRequestStartView {
+                        view: self.view.get(),
                         namespace: self.namespace,
                     });
                 }
@@ -1845,24 +1849,24 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
     /// election's `StartView` adopts this replica first). The replica sits
     /// in `Status::Recovering` meanwhile: it acks nothing, votes in no
     /// election, and initiates nothing.
-    pub fn begin_view_probe(&self) -> Vec<VsrAction> {
+    /// Returns nothing: the first probe rides the
+    /// `RequestStartViewMessage` timeout (~1s after boot), by which point
+    /// the replica mesh -- absent entirely at the boot-time call sites --
+    /// has formed. Emitting an action here implied a send that never
+    /// happened.
+    pub fn begin_view_probe(&self) {
         tracing::info!(
             replica = self.replica,
             namespace_raw = self.namespace,
             "beginning view probe"
         );
         self.status.set(Status::Recovering);
         self.probe_attempts.set(0);
-        {
-            let mut timeouts = self.timeouts.borrow_mut();
-            timeouts.stop(TimeoutKind::Prepare);
-            timeouts.stop(TimeoutKind::CommitMessage);
-            timeouts.stop(TimeoutKind::NormalHeartbeat);
-            timeouts.start(TimeoutKind::RequestStartViewMessage);
-        }
-        vec![VsrAction::SendRequestStartView {
-            namespace: self.namespace,
-        }]
+        let mut timeouts = self.timeouts.borrow_mut();
+        timeouts.stop(TimeoutKind::Prepare);
+        timeouts.stop(TimeoutKind::CommitMessage);
+        timeouts.stop(TimeoutKind::NormalHeartbeat);
+        timeouts.start(TimeoutKind::RequestStartViewMessage);
     }
 
     /// Peer side of the probe (sent by a Recovering replica at boot, or by
@@ -1893,7 +1897,17 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
             return Vec::new();
         }
         if self.primary_index(self.view.get()) == header.replica {
-            return self.start_election(plane, ViewChangeReason::PrimaryProbedView);
+            // Probes are re-broadcast on a timer, so delayed duplicates are
+            // the normal case, and `primary_index` is view % replica_count:
+            // a stale probe from replica R re-matches every replica_count
+            // views. Only a probe stamped with the CURRENT view proves the
+            // current primary is the one probing; anything else falls
+            // through (a backup answers nothing, and the true primary of a
+            // newer view answers with its StartView).
+            if header.view == self.view.get() {
+                return self.start_election(plane, ViewChangeReason::PrimaryProbedView);
+            }
+            return Vec::new();
         }
         if !self.is_primary() || self.ceded_primaryship.get() {
             return Vec::new();

core/integration/tests/server/scenarios/reconnect_after_restart_scenario.rs

@@ -608,3 +608,107 @@ async fn poll_from_zero_until(
         sleep(Duration::from_millis(250)).await;
     }
 }
+
+/// Rejoin with a window larger than the peers' evicted ring
+/// (`EVICTED_RING_CAPACITY` = 4096 entries): the serving peer answers
+/// `RangeEvicted` for the front of the range and the commit floor must
+/// settle at its retention point, with the rejoiner's recovered segments
+/// standing in below it. This is the only scenario that exercises
+/// `RangeEvicted` delivery and the floor end to end -- everything else
+/// fits inside the ring. The node-0 disk-growth assert is the proof the
+/// floor engaged: without it the commit walk gap-stops below the frontier
+/// and no post-restart batch ever flushes on the rejoiner.
+pub async fn run_ring_overflow_rejoin(harness: &mut TestHarness) {
+    const RING_OVERFLOW_OPS: u32 = 4300;
+    const POST_RESTART_OPS: u32 = 50;
+
+    let setup_client = harness
+        .root_client()
+        .await
+        .expect("Failed to create setup client");
+    setup_client
+        .create_stream(STREAM_NAME)
+        .await
+        .expect("Failed to create stream");
+    setup_client
+        .create_topic(
+            &Identifier::named(STREAM_NAME).unwrap(),
+            TOPIC_NAME,
+            1,
+            Default::default(),
+            None,
+            IggyExpiry::NeverExpire,
+            MaxTopicSize::ServerDefault,
+        )
+        .await
+        .expect("Failed to create topic");
+    send_messages(&setup_client, "ring", RING_OVERFLOW_OPS).await;
+    drop(setup_client);
+    // Let the backups' commit walks flush; the rejoin window must start
+    // from durable segments, not from a racing in-memory tail.
+    sleep(Duration::from_secs(2)).await;
+
+    let partition_path = format!(
+        "{}/streams/0/topics/0/partitions/0",
+        harness.server().data_path().display()
+    );
+    let durable_before = partition_log_bytes(&partition_path);
+    assert!(
+        durable_before > 0,
+        "pre-restart flush must have landed on node 0"
+    );
+
+    harness
+        .restart_server()
+        .await
+        .expect("Failed to restart node 0");
+
+    let client = harness
+        .root_client()
+        .await
+        .expect("Failed to reconnect after restart");
+    let polled = poll_from_zero_until(&client, RING_OVERFLOW_OPS, Duration::from_secs(60)).await;
+    assert_eq!(
+        polled.len(),
+        RING_OVERFLOW_OPS as usize,
+        "all pre-restart messages must survive the over-ring rejoin"
+    );
+
+    send_messages(&client, "post", POST_RESTART_OPS).await;
+    let all = poll_from_zero_until(
+        &client,
+        RING_OVERFLOW_OPS + POST_RESTART_OPS,
+        Duration::from_secs(60),
+    )
+    .await;
+    assert_eq!(all.len(), (RING_OVERFLOW_OPS + POST_RESTART_OPS) as usize);
+
+    // The rejoiner's commit walk must cross the floor: post-restart batches
+    // flush on node 0 only if commit_min advanced through the repaired
+    // window, so local disk growth is the floor-path proof.
+    let deadline = std::time::Instant::now() + Duration::from_secs(30);
+    loop {
+        if partition_log_bytes(&partition_path) > durable_before {
+            break;
+        }
+        assert!(
+            std::time::Instant::now() < deadline,
+            "node 0 never flushed past its recovered durable end: the \
+             commit walk is gap-stopped (floor path did not engage)"
+        );
+        sleep(Duration::from_millis(250)).await;
+    }
+}
+
+/// Sum of the `.log` segment file sizes under a partition directory.
+fn partition_log_bytes(partition_path: &str) -> u64 {
+    let Ok(entries) = std::fs::read_dir(partition_path) else {
+        return 0;
+    };
+    entries
+        .filter_map(Result::ok)
+        .filter(|e| e.path().extension().is_some_and(|ext| ext == "log"))
+        .filter_map(|e| e.metadata().ok())
+        .map(|m| m.len())
+        .sum()
+}

core/integration/tests/server/specific.rs

@@ -119,6 +119,18 @@ async fn full_cluster_restart_recovers_and_serves(harness: &mut TestHarness) {
     reconnect_after_restart_scenario::run_full_cluster_restart(harness).await;
 }
 
+// vsr-only: exercises `RangeEvicted` + the commit floor, which only exist
+// on the replicated plane (the rejoin window exceeds the peers' evicted
+// ring, so journal repair alone cannot cover it).
+#[cfg(feature = "vsr")]
+#[iggy_harness(server(
+    partition.messages_required_to_save = "1",
+    partition.enforce_fsync = true
+))]
+async fn rejoin_window_exceeding_evicted_ring(harness: &mut TestHarness) {
+    reconnect_after_restart_scenario::run_ring_overflow_rejoin(harness).await;
+}
+
 #[iggy_harness(server(
     partition.messages_required_to_save = "1",
     partition.enforce_fsync = true