feat(server-ng): replica rejoin via view probe and message repair

core/binary_protocol/src/consensus/command.rs

@@ -45,6 +45,20 @@ pub enum Command2 {
     ReplicaHello = 14,
     ReplicaChallenge = 15,
     ReplicaFinish = 16,
+
+    // Replica recovery: a restarted replica asks for the current view's
+    // `StartView` instead of trusting stale local state; only that view's
+    // primary answers (with a targeted `StartView`).
+    RequestStartView = 17,
+
+    // Journal repair: fetch committed prepares a replica is missing (rejoin
+    // window or interior hole). `RepairPrepare` carries a journaled prepare
+    // verbatim; `RangeEvicted` is the honest answer when the serving peer no
+    // longer retains the front of the range.
+    RequestPrepares = 18,
+    RepairPrepare = 19,
+    RepairDone = 20,
+    RangeEvicted = 21,
 }
 
 // SAFETY: Command2 is #[repr(u8)] with no padding bytes.
@@ -55,7 +69,7 @@ unsafe impl CheckedBitPattern for Command2 {
     type Bits = u8;
 
     fn is_valid_bit_pattern(bits: &u8) -> bool {
-        *bits <= Self::ReplicaFinish as u8
+        *bits <= Self::RangeEvicted as u8
     }
 }
 
@@ -76,16 +90,16 @@ mod tests {
 
     #[test]
     fn replica_auth_commands_are_valid_bit_patterns() {
-        // Locks the is_valid_bit_pattern bump: 14/15/16 parse, 17 still rejects.
-        for command in 14u8..=16 {
+        // Locks the is_valid_bit_pattern bump: 14..=21 parse, 22 still rejects.
+        for command in 14u8..=21 {
             let mut buf: AVec<u8, ConstAlign<16>> = AVec::new(16);
             buf.resize(256, 0);
             buf[60] = command;
             assert!(bytemuck::checked::try_from_bytes::<GenericHeader>(&buf).is_ok());
         }
         let mut buf: AVec<u8, ConstAlign<16>> = AVec::new(16);
         buf.resize(256, 0);
-        buf[60] = 17;
+        buf[60] = 22;
         assert!(bytemuck::checked::try_from_bytes::<GenericHeader>(&buf).is_err());
     }
 }

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>;
@@ -687,6 +695,42 @@ impl ConsensusHeader for PrepareHeader {
     }
 }
 
+// RepairPrepareHeader - repair peer -> recovering replica (journal repair)
+
+/// A stored prepare served for journal repair.
+///
+/// Byte-identical to [`PrepareHeader`] except the command, so the recovering
+/// replica routes it to the fence-free repair ingest instead of live
+/// replication. The newtype keeps the frame typed as `RepairPrepare` across
+/// every parse; converting to a live `Prepare` happens once, at the apply
+/// site.
+#[repr(transparent)]
+#[derive(Debug, Clone, Copy, CheckedBitPattern, NoUninit)]
+pub struct RepairPrepareHeader(pub PrepareHeader);
+
+impl ConsensusHeader for RepairPrepareHeader {
+    const COMMAND: Command2 = Command2::RepairPrepare;
+    fn operation(&self) -> Operation {
+        self.0.operation
+    }
+    fn command(&self) -> Command2 {
+        self.0.command
+    }
+    fn size(&self) -> u32 {
+        self.0.size
+    }
+
+    fn validate(&self) -> Result<(), ConsensusError> {
+        if self.0.command != Command2::RepairPrepare {
+            return Err(ConsensusError::InvalidCommand {
+                expected: Command2::RepairPrepare,
+                found: self.0.command,
+            });
+        }
+        Ok(())
+    }
+}
+
 // PrepareOkHeader - replica -> primary (acknowledgement)
 
 /// Replica -> primary: acknowledge a Prepare.
@@ -1018,6 +1062,194 @@ impl ConsensusHeader for StartViewHeader {
     }
 }
 
+// RequestStartViewHeader - restarted replica asking for the current view
+
+/// Recovering replica -> all replicas: resend me the current `StartView`.
+///
+/// Header-only; only the current view's primary answers, with a targeted
+/// `StartView` (adoption is fenced by the receiver's view monotonicity and
+/// the sender-is-primary check, so no nonce is needed).
+#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
+#[repr(C)]
+pub struct RequestStartViewHeader {
+    pub checksum: u128,
+    pub checksum_body: u128,
+    pub cluster: u128,
+    pub size: u32,
+    pub view: u32,
+    pub release: u32,
+    pub command: Command2,
+    pub replica: u8,
+    pub reserved_frame: [u8; 66],
+
+    pub namespace: u64,
+    pub reserved: [u8; 120],
+}
+const _: () = {
+    assert!(size_of::<RequestStartViewHeader>() == HEADER_SIZE);
+    assert!(
+        offset_of!(RequestStartViewHeader, namespace)
+            == offset_of!(RequestStartViewHeader, reserved_frame) + size_of::<[u8; 66]>()
+    );
+    assert!(offset_of!(RequestStartViewHeader, reserved) + size_of::<[u8; 120]>() == HEADER_SIZE);
+};
+
+impl ConsensusHeader for RequestStartViewHeader {
+    const COMMAND: Command2 = Command2::RequestStartView;
+    fn operation(&self) -> Operation {
+        Operation::Reserved
+    }
+    fn command(&self) -> Command2 {
+        self.command
+    }
+    fn size(&self) -> u32 {
+        self.size
+    }
+
+    fn validate(&self) -> Result<(), ConsensusError> {
+        if self.command != Command2::RequestStartView {
+            return Err(ConsensusError::InvalidCommand {
+                expected: Command2::RequestStartView,
+                found: self.command,
+            });
+        }
+        if self.release != 0 {
+            return Err(ConsensusError::InvalidField(
+                "release must be 0".to_string(),
+            ));
+        }
+        Ok(())
+    }
+}
+
+// RequestPreparesHeader - ask a peer for a range of committed prepares
+
+/// Recovering/holed replica -> a Normal peer: request a repair stream.
+///
+/// Sent to the primary first. Asks for the journaled prepares in
+/// `[from_op, to_op]` for `namespace`. Header-only. The peer answers with
+/// `RepairPrepare` frames in op order, terminated by `RepairDone` or
+/// `RangeEvicted`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
+#[repr(C)]
+pub struct RequestPreparesHeader {
+    pub checksum: u128,
+    pub checksum_body: u128,
+    pub cluster: u128,
+    pub size: u32,
+    pub view: u32,
+    pub release: u32,
+    pub command: Command2,
+    pub replica: u8,
+    pub reserved_frame: [u8; 66],
+
+    pub nonce: u128,
+    pub from_op: u64,
+    pub to_op: u64,
+    pub namespace: u64,
+    pub reserved: [u8; 88],
+}
+const _: () = {
+    assert!(size_of::<RequestPreparesHeader>() == HEADER_SIZE);
+    assert!(
+        offset_of!(RequestPreparesHeader, nonce)
+            == offset_of!(RequestPreparesHeader, reserved_frame) + size_of::<[u8; 66]>()
+    );
+    assert!(offset_of!(RequestPreparesHeader, reserved) + size_of::<[u8; 88]>() == HEADER_SIZE);
+};
+
+impl ConsensusHeader for RequestPreparesHeader {
+    const COMMAND: Command2 = Command2::RequestPrepares;
+    fn operation(&self) -> Operation {
+        Operation::Reserved
+    }
+    fn command(&self) -> Command2 {
+        self.command
+    }
+    fn size(&self) -> u32 {
+        self.size
+    }
+
+    fn validate(&self) -> Result<(), ConsensusError> {
+        if self.command != Command2::RequestPrepares {
+            return Err(ConsensusError::InvalidCommand {
+                expected: Command2::RequestPrepares,
+                found: self.command,
+            });
+        }
+        if self.from_op == 0 || self.from_op > self.to_op {
+            return Err(ConsensusError::InvalidField(
+                "repair range must be non-empty and 1-based".to_string(),
+            ));
+        }
+        Ok(())
+    }
+}
+
+// RepairRangeReplyHeader - RepairDone / RangeEvicted terminators
+
+/// Serving peer -> requester: terminates a repair stream.
+///
+/// As `RepairDone`, `through_op` is the last op served. As `RangeEvicted`,
+/// `retained_from` is the peer's oldest retained op -- everything older must
+/// come from bulk state sync (phase 3). One layout serves both commands.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
+#[repr(C)]
+pub struct RepairRangeReplyHeader {
+    pub checksum: u128,
+    pub checksum_body: u128,
+    pub cluster: u128,
+    pub size: u32,
+    pub view: u32,
+    pub release: u32,
+    pub command: Command2,
+    pub replica: u8,
+    pub reserved_frame: [u8; 66],
+
+    pub nonce: u128,
+    /// `RepairDone`: last op served. `RangeEvicted`: oldest retained op.
+    pub op: u64,
+    pub namespace: u64,
+    pub reserved: [u8; 96],
+}
+const _: () = {
+    assert!(size_of::<RepairRangeReplyHeader>() == HEADER_SIZE);
+    assert!(
+        offset_of!(RepairRangeReplyHeader, nonce)
+            == offset_of!(RepairRangeReplyHeader, reserved_frame) + size_of::<[u8; 66]>()
+    );
+    assert!(offset_of!(RepairRangeReplyHeader, reserved) + size_of::<[u8; 96]>() == HEADER_SIZE);
+};
+
+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
+    }
+    fn command(&self) -> Command2 {
+        self.command
+    }
+    fn size(&self) -> u32 {
+        self.size
+    }
+
+    fn validate(&self) -> Result<(), ConsensusError> {
+        if self.command != Command2::RepairDone && self.command != Command2::RangeEvicted {
+            return Err(ConsensusError::InvalidCommand {
+                expected: Command2::RepairDone,
+                found: self.command,
+            });
+        }
+        Ok(())
+    }
+}
+
 // Tests
 
 #[cfg(test)]

core/binary_protocol/src/consensus/mod.rs

@@ -46,8 +46,10 @@ pub use command::Command2;
 pub use error::ConsensusError;
 pub use header::{
     CommitHeader, ConsensusHeader, DoViewChangeHeader, EvictionHeader, EvictionReason,
-    GenericHeader, HEADER_SIZE, PrepareHeader, PrepareOkHeader, RESERVED_COMMAND_LEN, ReplyHeader,
-    RequestHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader, StartViewHeader, read_size_field,
+    GenericHeader, HEADER_SIZE, PrepareHeader, PrepareOkHeader, RESERVED_COMMAND_LEN,
+    RepairPrepareHeader, RepairRangeReplyHeader, ReplyHeader, RequestHeader, RequestPreparesHeader,
+    RequestStartViewHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader, StartViewHeader,
+    read_size_field,
 };
 pub use operation::Operation;
 pub use reply_result::{RESULT_COUNT_LEN, RESULT_ENTRY_LEN, result_code, result_section_len};

core/binary_protocol/src/lib.rs

@@ -73,7 +73,8 @@ pub use codec::{WireDecode, WireEncode};
 pub use consensus::{
     Command2, CommitHeader, ConsensusError, ConsensusHeader, DoViewChangeHeader, EvictionHeader,
     EvictionReason, GenericHeader, HEADER_SIZE, Operation, PrepareHeader, PrepareOkHeader,
-    RESERVED_COMMAND_LEN, ReplyHeader, RequestHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader,
+    RESERVED_COMMAND_LEN, RepairPrepareHeader, RepairRangeReplyHeader, ReplyHeader, RequestHeader,
+    RequestPreparesHeader, RequestStartViewHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader,
     StartViewHeader, read_size_field, result_code, result_section_len,
 };
 pub use dispatch::{COMMAND_TABLE, CommandMeta, lookup_by_operation, lookup_command};