fix(cluster)!: let slow replicas catch up under load

core/binary_protocol/src/consensus/command.rs

@@ -96,6 +96,8 @@ pub enum Command {
     // Fresh metadata and partition readiness for primary rebalancing.
     PrimaryRebalance = 33,
     SegmentDelete = 34,
+    CommitReply = 35,
+    PrepareProgress = 36,
 }
 
 // SAFETY: Command is #[repr(u8)] with no padding bytes.
@@ -106,7 +108,7 @@ unsafe impl CheckedBitPattern for Command {
     type Bits = u8;
 
     fn is_valid_bit_pattern(bits: &u8) -> bool {
-        *bits <= Self::SegmentDelete as u8
+        *bits <= Self::PrepareProgress as u8
     }
 }
 
@@ -127,15 +129,15 @@ mod tests {
 
     #[test]
     fn replica_auth_commands_are_valid_bit_patterns() {
-        for command in 14u8..=34 {
+        for command in 14u8..=36 {
             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] = 35;
+        buf[60] = 37;
         assert!(bytemuck::checked::try_from_bytes::<GenericHeader>(&buf).is_err());
     }
 }

core/binary_protocol/src/consensus/header.rs

@@ -1499,6 +1499,166 @@ impl ConsensusHeader for CommitHeader {
     }
 }
 
+/// Backup -> primary: echo a heartbeat counter for quorum liveness.
+///
+/// This acknowledges connectivity, not journal or state-machine durability.
+/// Nodes using this message require a coordinated cluster upgrade.
+#[repr(C)]
+#[derive(Debug, Clone, Copy, CheckedBitPattern, NoUninit)]
+pub struct CommitReplyHeader {
+    pub checksum: u128,
+    pub checksum_body: u128,
+    pub cluster: u128,
+    pub size: u32,
+    pub view: u32,
+    pub release: u32,
+    pub command: Command,
+    pub replica: u8,
+    pub reserved_frame: [u8; 66],
+    pub timestamp_monotonic: u64,
+    pub group: u64,
+    pub partition_incarnation: u64,
+    pub reserved: [u8; 104],
+}
+
+const _: () = {
+    assert!(size_of::<CommitReplyHeader>() == HEADER_SIZE);
+    assert!(offset_of!(CommitReplyHeader, timestamp_monotonic) == 128);
+    assert!(offset_of!(CommitReplyHeader, partition_incarnation) == 144);
+    assert!(offset_of!(CommitReplyHeader, reserved) + size_of::<[u8; 104]>() == HEADER_SIZE);
+};
+
+impl ConsensusHeader for CommitReplyHeader {
+    const FRAME_SEALED: bool = true;
+    const COMMAND: Command = Command::CommitReply;
+
+    fn checksum(&self) -> u128 {
+        self.checksum
+    }
+    fn set_checksum(&mut self, checksum: u128) {
+        self.checksum = checksum;
+    }
+    fn operation(&self) -> Operation {
+        Operation::Reserved
+    }
+    fn command(&self) -> Command {
+        self.command
+    }
+    fn size(&self) -> u32 {
+        self.size
+    }
+
+    fn validate(&self) -> Result<(), ConsensusError> {
+        if self.command != Self::COMMAND {
+            return Err(ConsensusError::InvalidCommand {
+                expected: Self::COMMAND,
+                found: self.command,
+            });
+        }
+        if usize::try_from(self.size) != Ok(HEADER_SIZE) {
+            return Err(ConsensusError::InvalidField(
+                "commit_reply: invalid size".to_string(),
+            ));
+        }
+        if self.release != 0 {
+            return Err(ConsensusError::InvalidField("release != 0".to_string()));
+        }
+        Ok(())
+    }
+}
+
+/// Correlates a replication checkpoint with the backup's contiguous prefixes.
+///
+/// Probes must stay ordered with prepares on the sender's queue, so a reply
+/// proves that the backup consumed every frame sent before its probe. That FIFO
+/// position, not an op, releases transport credit. Progress never counts toward
+/// a commit quorum.
+/// Replies echo the nonce and observed op. `held_op` is the end of the prefix
+/// that the backup holds with proof in the current view, so the primary finds
+/// replication gaps from it. `received_op <= held_op` is the part of that
+/// prefix that meets the prepare acknowledgment rules, and `ready` certifies it. Probes advertise the primary's applied frontier for retention
+/// fallback.
+/// Nodes using this message require a coordinated cluster upgrade.
+#[repr(C)]
+#[derive(Debug, Clone, Copy, CheckedBitPattern, NoUninit)]
+pub struct PrepareProgressHeader {
+    pub checksum: u128,
+    pub checksum_body: u128,
+    pub cluster: u128,
+    pub size: u32,
+    pub view: u32,
+    pub release: u32,
+    pub command: Command,
+    pub replica: u8,
+    pub reserved_frame: [u8; 66],
+    pub nonce: u64,
+    pub group: u64,
+    pub partition_incarnation: u64,
+    pub observed_op: u64,
+    pub received_op: u64,
+    pub held_op: u64,
+    pub commit_op: u64,
+    pub kind: u8,
+    pub ready: u8,
+    pub reserved: [u8; 70],
+}
+
+const _: () = {
+    assert!(size_of::<PrepareProgressHeader>() == HEADER_SIZE);
+    assert!(offset_of!(PrepareProgressHeader, nonce) == 128);
+    assert!(offset_of!(PrepareProgressHeader, partition_incarnation) == 144);
+    assert!(offset_of!(PrepareProgressHeader, reserved) + size_of::<[u8; 70]>() == HEADER_SIZE);
+};
+
+impl ConsensusHeader for PrepareProgressHeader {
+    const FRAME_SEALED: bool = true;
+    const COMMAND: Command = Command::PrepareProgress;
+
+    fn checksum(&self) -> u128 {
+        self.checksum
+    }
+    fn set_checksum(&mut self, checksum: u128) {
+        self.checksum = checksum;
+    }
+    fn operation(&self) -> Operation {
+        Operation::Reserved
+    }
+    fn command(&self) -> Command {
+        self.command
+    }
+    fn size(&self) -> u32 {
+        self.size
+    }
+
+    fn validate(&self) -> Result<(), ConsensusError> {
+        if self.command != Self::COMMAND {
+            return Err(ConsensusError::InvalidCommand {
+                expected: Self::COMMAND,
+                found: self.command,
+            });
+        }
+        if usize::try_from(self.size) != Ok(HEADER_SIZE) {
+            return Err(ConsensusError::InvalidField(
+                "prepare_progress: invalid size".to_string(),
+            ));
+        }
+        if self.release != 0 {
+            return Err(ConsensusError::InvalidField("release != 0".to_string()));
+        }
+        if self.kind > Self::REPLY || self.ready > 1 {
+            return Err(ConsensusError::InvalidField(
+                "prepare_progress: invalid fields".to_string(),
+            ));
+        }
+        Ok(())
+    }
+}
+
+impl PrepareProgressHeader {
+    pub const PROBE: u8 = 0;
+    pub const REPLY: u8 = 1;
+}
+
 // StartViewChangeHeader - failure detection (header-only)
 
 /// Replica suspects primary failure. Header-only.
@@ -3156,13 +3316,14 @@ mod tests {
     }
     use super::prepare_identity_checksum_bytes;
     use super::{
-        Command, CommitHeader, ConsensusError, ConsensusHeader, DoViewChangeHeader, EvictionHeader,
-        EvictionReason, ForwardCompletionHeader, ForwardLogoutHeader, ForwardLogoutOutcome,
-        ForwardLogoutResultHeader, ForwardRegisterHeader, ForwardRegisterOutcome,
-        ForwardRegisterResultHeader, GenericHeader, HEADER_SIZE, LIFECYCLE_BATCH_HEADER_SIZE,
-        LIFECYCLE_COMPLETION_SIZE, LIFECYCLE_COMPLETIONS_MAX, Operation, PrepareHeader,
-        PrepareOkHeader, RepairPrepareHeader, RepairRangeReplyHeader, ReplyHeader, RequestHeader,
-        RequestPreparesHeader, RequestStartViewHeader, RequestStateChunkHeader,
+        Command, CommitHeader, CommitReplyHeader, ConsensusError, ConsensusHeader,
+        DoViewChangeHeader, EvictionHeader, EvictionReason, ForwardCompletionHeader,
+        ForwardLogoutHeader, ForwardLogoutOutcome, ForwardLogoutResultHeader,
+        ForwardRegisterHeader, ForwardRegisterOutcome, ForwardRegisterResultHeader, GenericHeader,
+        HEADER_SIZE, LIFECYCLE_BATCH_HEADER_SIZE, LIFECYCLE_COMPLETION_SIZE,
+        LIFECYCLE_COMPLETIONS_MAX, Operation, PrepareHeader, PrepareOkHeader,
+        PrepareProgressHeader, RepairPrepareHeader, RepairRangeReplyHeader, ReplyHeader,
+        RequestHeader, RequestPreparesHeader, RequestStartViewHeader, RequestStateChunkHeader,
         RequestStateTransferHeader, RoutedRequestHeader, StartViewChangeHeader, StartViewHeader,
         StateChunkHeader, StateTransferTargetHeader,
     };
@@ -3226,6 +3387,8 @@ mod tests {
         seals!(
             PrepareOkHeader,
             CommitHeader,
+            CommitReplyHeader,
+            PrepareProgressHeader,
             StartViewChangeHeader,
             DoViewChangeHeader,
             StartViewHeader,
@@ -3292,6 +3455,8 @@ mod tests {
         }
         check_incarnation!(
             CommitHeader,
+            CommitReplyHeader,
+            PrepareProgressHeader,
             StartViewChangeHeader,
             DoViewChangeHeader,
             StartViewHeader,

core/binary_protocol/src/consensus/mod.rs

@@ -52,15 +52,16 @@ pub use consumer_session::{
 };
 pub use error::ConsensusError;
 pub use header::{
-    CHECKSUM_UNSEALED, CommitHeader, ConsensusHeader, DVC_HEADERS_MAX, DoViewChangeHeader,
-    EvictionHeader, EvictionReason, ForwardCompletionHeader, ForwardLogoutHeader,
-    ForwardLogoutOutcome, ForwardLogoutResultHeader, ForwardRegisterHeader, ForwardRegisterOutcome,
-    ForwardRegisterResultHeader, GenericHeader, HEADER_SIZE, LIFECYCLE_BATCH_HEADER_SIZE,
-    LIFECYCLE_COMPLETION_SIZE, LIFECYCLE_COMPLETIONS_MAX, PrepareHeader, PrepareOkHeader,
-    RESERVED_COMMAND_LEN, RepairPrepareHeader, RepairRangeReplyHeader, ReplyHeader, RequestHeader,
-    RequestPreparesHeader, RequestStartViewHeader, RequestStateChunkHeader,
-    RequestStateTransferHeader, RoutedRequestHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader,
-    StartViewHeader, StateChunkHeader, StateTransferTargetHeader, frame_body, frame_checksum_bytes,
+    CHECKSUM_UNSEALED, CommitHeader, CommitReplyHeader, ConsensusHeader, DVC_HEADERS_MAX,
+    DoViewChangeHeader, EvictionHeader, EvictionReason, ForwardCompletionHeader,
+    ForwardLogoutHeader, ForwardLogoutOutcome, ForwardLogoutResultHeader, ForwardRegisterHeader,
+    ForwardRegisterOutcome, ForwardRegisterResultHeader, GenericHeader, HEADER_SIZE,
+    LIFECYCLE_BATCH_HEADER_SIZE, LIFECYCLE_COMPLETION_SIZE, LIFECYCLE_COMPLETIONS_MAX,
+    PrepareHeader, PrepareOkHeader, PrepareProgressHeader, RESERVED_COMMAND_LEN,
+    RepairPrepareHeader, RepairRangeReplyHeader, ReplyHeader, RequestHeader, RequestPreparesHeader,
+    RequestStartViewHeader, RequestStateChunkHeader, RequestStateTransferHeader,
+    RoutedRequestHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader, StartViewHeader,
+    StateChunkHeader, StateTransferTargetHeader, frame_body, frame_checksum_bytes,
     prepare_identity_checksum_bytes, read_size_field,
 };
 pub use operation::Operation;

core/binary_protocol/src/lib.rs

@@ -70,20 +70,20 @@ pub mod version;
 
 pub use codec::{WireDecode, WireEncode};
 pub use consensus::{
-    CHECKSUM_UNSEALED, Command, CommitHeader, ConsensusError, ConsensusHeader, ConsumerSession,
-    ConsumerSessionHeartbeatHeader, DVC_HEADERS_MAX, DoViewChangeHeader, EvictionHeader,
-    EvictionReason, ForwardCompletionHeader, ForwardLogoutHeader, ForwardLogoutOutcome,
-    ForwardLogoutResultHeader, ForwardRegisterHeader, ForwardRegisterOutcome,
+    CHECKSUM_UNSEALED, Command, CommitHeader, CommitReplyHeader, ConsensusError, ConsensusHeader,
+    ConsumerSession, ConsumerSessionHeartbeatHeader, DVC_HEADERS_MAX, DoViewChangeHeader,
+    EvictionHeader, EvictionReason, ForwardCompletionHeader, ForwardLogoutHeader,
+    ForwardLogoutOutcome, ForwardLogoutResultHeader, ForwardRegisterHeader, ForwardRegisterOutcome,
     ForwardRegisterResultHeader, GenericHeader, HEADER_SIZE, LIFECYCLE_BATCH_HEADER_SIZE,
     LIFECYCLE_COMPLETION_SIZE, LIFECYCLE_COMPLETIONS_MAX, MAX_CONSUMER_SESSIONS_PER_HEARTBEAT,
-    Operation, PrepareHeader, PrepareOkHeader, PrimaryRebalanceHeader, PrimaryRebalanceKind,
-    REJECTION_SECTION_LEN, RESERVED_COMMAND_LEN, RepairPrepareHeader, RepairRangeReplyHeader,
-    ReplyHeader, RequestHeader, RequestPreparesHeader, RequestStartViewHeader,
-    RequestStateChunkHeader, RequestStateTransferHeader, RoutedRequestHeader, SIZE_FIELD_OFFSET,
-    SegmentDeleteHeader, SegmentDeleteKind, StartViewChangeHeader, StartViewHeader,
-    StateChunkHeader, StateTransferTargetHeader, frame_body, frame_checksum_bytes,
-    prepare_identity_checksum_bytes, read_size_field, result_code, result_section_len,
-    write_rejection_section,
+    Operation, PrepareHeader, PrepareOkHeader, PrepareProgressHeader, PrimaryRebalanceHeader,
+    PrimaryRebalanceKind, REJECTION_SECTION_LEN, RESERVED_COMMAND_LEN, RepairPrepareHeader,
+    RepairRangeReplyHeader, ReplyHeader, RequestHeader, RequestPreparesHeader,
+    RequestStartViewHeader, RequestStateChunkHeader, RequestStateTransferHeader,
+    RoutedRequestHeader, SIZE_FIELD_OFFSET, SegmentDeleteHeader, SegmentDeleteKind,
+    StartViewChangeHeader, StartViewHeader, StateChunkHeader, StateTransferTargetHeader,
+    frame_body, frame_checksum_bytes, prepare_identity_checksum_bytes, read_size_field,
+    result_code, result_section_len, write_rejection_section,
 };
 pub use dispatch::{
     COMMAND_TABLE, CommandMeta, command_index, lookup_by_operation, lookup_command,

core/configs/src/server_config/cluster.rs

@@ -200,9 +200,9 @@ fn default_superblock_wedged_fatal_timeout() -> IggyDuration {
 pub struct ClusterConfig {
     pub enabled: bool,
     pub name: String,
-    /// Backup-side liveness window for a plane's primary. A replica that
-    /// sees no primary traffic for this long starts a view change
-    /// (`normal_heartbeat_timeout`). Raise it on oversubscribed hosts where
+    /// Liveness window for the primary and its replication quorum. Backups
+    /// vote after primary silence; a primary without heartbeat replies from
+    /// a quorum relinquishes leadership. Raise it on oversubscribed hosts where
     /// scheduling stalls fake primary death; sub-`MIN_CLUSTER_HEARTBEAT_TIMEOUT`
     /// values (including the `0` / `disabled` / `unlimited` sentinels, which
     /// all parse to zero) are rejected at boot. On Linux it is also the

core/configs/src/server_config/defaults.rs

@@ -193,6 +193,10 @@ impl Default for PartitionConfig {
                 .wal_bytes_max
                 .parse()
                 .expect("embedded WAL capacity is valid"),
+            repair_memory_bytes_max: partition
+                .repair_memory_bytes_max
+                .parse()
+                .expect("embedded repair memory capacity is valid"),
             wal_group_commit_delay_micros: u64::try_from(partition.wal_group_commit_delay_micros)
                 .expect("embedded WAL group commit delay is valid"),
             validate_checksum: SERVER_CONFIG.partition.validate_checksum,