diff --git a/core/binary_protocol/src/consensus/command.rs b/core/binary_protocol/src/consensus/command.rs index 6adafef83e..d2d0ada7cc 100644 --- a/core/binary_protocol/src/consensus/command.rs +++ b/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,8 +90,8 @@ 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> = AVec::new(16); buf.resize(256, 0); buf[60] = command; @@ -85,7 +99,7 @@ mod tests { } let mut buf: AVec> = AVec::new(16); buf.resize(256, 0); - buf[60] = 17; + buf[60] = 22; assert!(bytemuck::checked::try_from_bytes::(&buf).is_err()); } } diff --git a/core/binary_protocol/src/consensus/header.rs b/core/binary_protocol/src/consensus/header.rs index 3913f5f58e..7c852e150e 100644 --- a/core/binary_protocol/src/consensus/header.rs +++ b/core/binary_protocol/src/consensus/header.rs @@ -70,6 +70,14 @@ pub fn read_size_field(header: &[u8]) -> Option { 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::() == 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::() == 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::() == 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)] diff --git a/core/binary_protocol/src/consensus/mod.rs b/core/binary_protocol/src/consensus/mod.rs index ddfab33561..594d1ed989 100644 --- a/core/binary_protocol/src/consensus/mod.rs +++ b/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}; diff --git a/core/binary_protocol/src/lib.rs b/core/binary_protocol/src/lib.rs index b16fe8c6cd..1d2010f3f6 100644 --- a/core/binary_protocol/src/lib.rs +++ b/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}; diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index e38d3ddc54..ee9ec60164 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -26,7 +26,7 @@ use crate::{ use bit_set::BitSet; use iggy_binary_protocol::{ Command2, ConsensusHeader, DoViewChangeHeader, GenericHeader, PrepareHeader, PrepareOkHeader, - ReplyHeader, RequestHeader, StartViewChangeHeader, StartViewHeader, + ReplyHeader, RequestHeader, RequestStartViewHeader, StartViewChangeHeader, StartViewHeader, }; use iggy_common::IggyTimestamp; use message_bus::IggyMessageBus; @@ -101,6 +101,11 @@ pub const PIPELINE_REQUEST_QUEUE_MAX: usize = 64; /// Maximum number of replicas in a cluster. pub const REPLICAS_MAX: usize = 32; +/// Unanswered `RequestStartView` probes tolerated before a recovering +/// replica gives up waiting for a settled primary and falls back to an +/// election (a full-cluster restart leaves nobody able to answer). +pub const PROBE_ATTEMPTS_MAX: u32 = 5; + /// Maximum number of clients tracked in the clients table. /// When exceeded, the client with the oldest committed request is evicted. pub const CLIENTS_TABLE_MAX: usize = 8192; @@ -573,6 +578,11 @@ pub enum VsrAction { commit: u64, namespace: u64, }, + /// Broadcast a `RequestStartView` probe (recovering replica asking for + /// the current view's `StartView`; only that view's primary answers). + /// 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, @@ -669,13 +679,6 @@ where /// Commit point the recovered WAL suffix must re-reach before admitting /// client requests as primary (`0` = no recovered suffix pending). recovery_barrier: Cell, - /// True until a replica that booted without its consensus state (see - /// `init_recovering`) learns the cluster commit point. While set, the - /// first `StartView` / `Commit` fast-forwards `commit_min` to the learned - /// `commit_max`: the replica's recovered durable state stands in for the - /// journal prefix it no longer has, so walking `commit_journal` from op 1 - /// would find nothing and declare divergence. - recovering: Cell, /// True while this replica declines the primaryship its (stale) recovered /// view assigns it (see `init_as_backup`). `is_primary()` is pure view /// math, so without this flag a restarted view-N primary would still pass @@ -706,6 +709,9 @@ where loopback_queue: RefCell>>, /// Tracks start view change messages received from all replicas (including self) start_view_change_from_all_replicas: RefCell>, + /// Consecutive unanswered `RequestStartView` probes while Recovering; + /// at [`PROBE_ATTEMPTS_MAX`] the replica falls back to an election. + probe_attempts: Cell, /// Tracks DVC messages received (only used by primary candidate) /// Stores metadata; actual log comes from message @@ -763,7 +769,6 @@ impl> VsrConsensus { view: Cell::new(0), log_view: Cell::new(0), recovery_barrier: Cell::new(0), - recovering: Cell::new(false), ceded_primaryship: Cell::new(false), status: Cell::new(Status::Recovering), sequencer: LocalSequencer::new(0), @@ -775,6 +780,7 @@ impl> VsrConsensus { message_bus, loopback_queue: RefCell::new(VecDeque::with_capacity(PIPELINE_PREPARE_QUEUE_MAX)), start_view_change_from_all_replicas: RefCell::new(BitSet::with_capacity(REPLICAS_MAX)), + probe_attempts: Cell::new(0), do_view_change_from_all_replicas: RefCell::new(dvc_quorum_array_empty()), do_view_change_quorum: Cell::new(false), sent_own_start_view_change: Cell::new(false), @@ -818,18 +824,6 @@ impl> VsrConsensus { self.ceded_primaryship.get() } - /// Initialize a replica whose consensus state did NOT survive restart - /// (e.g. a partition group: its journal is in-memory and its segments - /// carry no op numbers). Such a replica must not assume primaryship: it - /// would heartbeat `commit_min = 0`, dragging the group behind backups - /// that kept their journals, and it has no ops to re-pipeline. Join as a - /// backup instead; the peers' heartbeat timeout elects a primary that - /// still holds the log, and its `StartView` brings this replica forward. - pub fn init_recovering(&self) { - self.init_as_backup(); - self.recovering.set(true); - } - #[must_use] // cast_lossless: `u32::from()` unavailable in const fn. // cast_possible_truncation: modulo by replica_count (u8) guarantees result fits in u8. @@ -1209,6 +1203,67 @@ impl> VsrConsensus { timeouts = self.timeouts.borrow_mut(); } + if timeouts.fired(TimeoutKind::RequestStartViewMessage) { + drop(timeouts); + // Two probers share this timeout, both asking "resend me the + // current StartView": + // - Recovering (boot probe): re-broadcast until the settled + // primary answers or an election's StartView adopts us. + // - ViewChange backup: the election may have concluded with our + // copy of the StartView lost; re-requesting it is a + // two-message fix, while the ViewChangeStatus escalation + // backstop burns a fresh cluster-wide election. The would-be + // primary of the view skips the probe (it concludes the view + // itself or escalates). + match self.status.get() { + Status::Recovering => { + // A probe answered by nobody, repeatedly, means nobody is + // settled -- the whole cluster restarted together and + // every group sits quorum-invisible waiting for a primary + // that cannot exist. Fall back to an election: recovered + // WALs compete on (log_view, op) in the DVC exchange, so + // the best surviving log leads; a group whose members all + // rejoined journal-less elects on equal terms and stands + // on its recovered durable state. Any still-live settled + // primary answers well before the fallback fires. + let attempts = self.probe_attempts.get() + 1; + self.probe_attempts.set(attempts); + if attempts >= PROBE_ATTEMPTS_MAX { + self.finish_view_probe(); + actions.extend( + self.start_election(plane, ViewChangeReason::ViewProbeUnanswered), + ); + } else { + self.timeouts + .borrow_mut() + .reset(TimeoutKind::RequestStartViewMessage); + actions.push(VsrAction::SendRequestStartView { + view: self.view.get(), + namespace: self.namespace, + }); + } + } + Status::ViewChange if self.primary_index(self.view.get()) != self.replica => { + self.timeouts + .borrow_mut() + .reset(TimeoutKind::RequestStartViewMessage); + actions.push(VsrAction::SendRequestStartView { + view: self.view.get(), + namespace: self.namespace, + }); + } + _ => { + // Stale arm (e.g. went Normal without passing an exit + // that stops it): silence it instead of refiring every + // tick. + self.timeouts + .borrow_mut() + .stop(TimeoutKind::RequestStartViewMessage); + } + } + timeouts = self.timeouts.borrow_mut(); + } + if timeouts.fired(TimeoutKind::ViewChangeStatus) { drop(timeouts); actions.extend(self.handle_view_change_status_timeout(plane)); @@ -1221,6 +1276,12 @@ impl> VsrConsensus { /// Called when `normal_heartbeat` timeout fires. /// Backup hasn't heard from primary - start view change. fn handle_normal_heartbeat_timeout(&self, plane: PlaneKind) -> Vec { + // A recovering replica makes progress through RequestStartView + // retries, not elections; it is quorum-invisible. + if self.status.get() == Status::Recovering { + return Vec::new(); + } + // Only backups trigger view change on heartbeat timeout. `is_primary` // is pure view math though: a replica that booted recovering / with // ceded primaryship while sitting at the primary index is a backup by @@ -1228,7 +1289,7 @@ impl> VsrConsensus { // start an election, silently dropping out of quorum until an // unrelated view change rescues it. Let it climb StartViewChange like // any other backup. - if self.is_primary() && !self.recovering.get() && !self.ceded_primaryship.get() { + if self.is_primary() && !self.ceded_primaryship.get() { return Vec::new(); } @@ -1237,7 +1298,11 @@ impl> VsrConsensus { return Vec::new(); } - // Advance to new view and transition to view change + self.start_election(plane, ViewChangeReason::NormalHeartbeatTimeout) + } + + /// Advance to `view + 1` and start a view change (own SVC counted). + fn start_election(&self, plane: PlaneKind, reason: ViewChangeReason) -> Vec { let old_view = self.view.get(); let new_view = old_view + 1; @@ -1249,12 +1314,12 @@ impl> VsrConsensus { .borrow_mut() .insert(self.replica as usize); - // Update timeouts for view change status { let mut timeouts = self.timeouts.borrow_mut(); timeouts.stop(TimeoutKind::NormalHeartbeat); timeouts.start(TimeoutKind::StartViewChangeMessage); timeouts.start(TimeoutKind::ViewChangeStatus); + timeouts.start(TimeoutKind::RequestStartViewMessage); } emit_sim_event( @@ -1263,7 +1328,7 @@ impl> VsrConsensus { replica: ReplicaLogContext::from_consensus(self, plane), old_view, new_view, - reason: ViewChangeReason::NormalHeartbeatTimeout, + reason, }, ); @@ -1420,19 +1485,17 @@ impl> VsrConsensus { /// Only acts on the primary in normal status with a non-empty pipeline. /// Resets the timeout with backoff on each firing. fn handle_prepare_timeout(&self) -> Vec { - // TODO(prepare-timeout): adopt TigerBeetle's timer lifecycle - // (replica.zig `on_prepare_ok` / `on_prepare_timeout`). They - // disarm in the ack path the moment quorum drains the pipeline - // (`stop()`) and rearm for the next-oldest prepare when one - // commits with others still pending (`reset()`), giving the - // invariant "ticking iff pipeline non-empty" (asserted in their - // timeout handler) and a timeout that always measures the - // current oldest prepare's age. Ours arms once per idle->busy - // transition and disarms lazily below, so a prepare pushed late - // into an armed window can be retransmitted before it is - // `PREPARE_TICKS` old. They also special-case "all remote acks - // present, own journal write is the laggard" by retrying the - // local write instead of retransmitting. + // TODO(prepare-timeout): tighten the timer lifecycle: disarm in + // the ack path the moment quorum drains the pipeline and rearm + // for the next-oldest prepare when one commits with others still + // pending, giving the invariant "ticking iff pipeline non-empty" + // and a timeout that always measures the current oldest + // prepare's age. Ours arms once per idle->busy transition and + // disarms lazily below, so a prepare pushed late into an armed + // window can be retransmitted before it is `PREPARE_TICKS` old. + // Also worth special-casing "all remote acks present, own + // journal write is the laggard" by retrying the local write + // instead of retransmitting. // // Every early return below must stop or back off the timeout. // `fired()` stays true until the timer is rearmed, so returning @@ -1511,6 +1574,13 @@ impl> VsrConsensus { header.namespace, self.namespace, "SVC routed to wrong group" ); + // A recovering replica is quorum-invisible: it lost (or cannot trust) + // its durable state, so it must not vote history into existence. The + // election proceeds among the peers; its conclusion reaches this + // replica via StartView, which recovery accepts. + if self.status.get() == Status::Recovering { + return Vec::new(); + } let from_replica = header.replica; let msg_view = header.view; @@ -1538,6 +1608,7 @@ impl> VsrConsensus { timeouts.stop(TimeoutKind::NormalHeartbeat); timeouts.start(TimeoutKind::StartViewChangeMessage); timeouts.start(TimeoutKind::ViewChangeStatus); + timeouts.start(TimeoutKind::RequestStartViewMessage); } emit_sim_event( @@ -1659,6 +1730,11 @@ impl> VsrConsensus { header.namespace, self.namespace, "DVC routed to wrong group" ); + // Quorum-invisible while recovering (see handle_start_view_change): + // a recovering replica must not collect DVCs and crown itself. + if self.status.get() == Status::Recovering { + return Vec::new(); + } let from_replica = header.replica; let msg_view = header.view; let msg_log_view = header.log_view; @@ -1689,6 +1765,7 @@ impl> VsrConsensus { timeouts.stop(TimeoutKind::NormalHeartbeat); timeouts.start(TimeoutKind::StartViewChangeMessage); timeouts.start(TimeoutKind::ViewChangeStatus); + timeouts.start(TimeoutKind::RequestStartViewMessage); } emit_sim_event( @@ -1766,6 +1843,113 @@ impl> VsrConsensus { actions } + /// Begin the view probe: broadcast `RequestStartView` and keep + /// re-broadcasting on `TimeoutKind::RequestStartViewMessage` until the + /// current view's primary answers with a targeted `StartView` (or an + /// election's `StartView` adopts this replica first). The replica sits + /// in `Status::Recovering` meanwhile: it acks nothing, votes in no + /// election, and initiates nothing. + /// 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); + } + + /// Peer side of the probe (sent by a Recovering replica at boot, or by + /// a `ViewChange` backup whose copy of the concluding `StartView` was + /// lost). Only the current view's PRIMARY answers, with a `StartView`; + /// backups stay silent and the prober retries. Special case: a probe + /// FROM the replica that is the current view's primary-by-index proves + /// that primary cannot lead (a probing replica has either lost its + /// state or abandoned the view), so a peer receiving it elects + /// immediately instead of waiting out the heartbeat timeout on a slot + /// known to be dead. + /// + /// # Panics + /// Panics when the probe is routed to the wrong group. + pub fn handle_request_start_view( + &self, + plane: PlaneKind, + header: &RequestStartViewHeader, + ) -> Vec { + assert_eq!( + header.namespace, self.namespace, + "RequestStartView routed to wrong group" + ); + if self.status.get() != Status::Normal { + return Vec::new(); + } + if header.replica == self.replica { + return Vec::new(); + } + if self.primary_index(self.view.get()) == header.replica { + // 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(); + } + // A primary mid-transition (log_view lagging) has no settled + // frontier to publish yet. + if self.log_view.get() != self.view.get() { + return Vec::new(); + } + vec![VsrAction::SendStartView { + view: self.view.get(), + op: self.sequencer.current_sequence(), + commit: self.commit_max.get(), + namespace: self.namespace, + }] + } + + /// Set the commit floor after journal repair filled `(floor, commit_max]`: + /// everything at or below `floor` is represented by this replica's + /// recovered durable state (segments + offset files), proven by the + /// serving peer answering `RangeEvicted { retained_from = floor + 1 }`. + /// Unlike the retired first-commit fast-forward, ops in the repair window + /// are journaled and WALKED, never skipped. + /// + /// # Panics + /// Panics when `floor` would rewind the already-executed `commit_min`. + pub fn set_commit_floor(&self, floor: u64) { + let current = self.commit_min.get(); + assert!( + current <= floor, + "commit floor {floor} may not rewind commit_min {current}" + ); + self.commit_min.set(floor); + } + + fn finish_view_probe(&self) { + self.probe_attempts.set(0); + self.timeouts + .borrow_mut() + .stop(TimeoutKind::RequestStartViewMessage); + } + /// Handle a received `StartView` message (backups only). /// /// "When other replicas receive the STARTVIEW message, they replace their log @@ -1823,12 +2007,15 @@ impl> VsrConsensus { commit = msg_commit, "adopting view from StartView" ); + // A StartView concluding around an in-flight view probe supersedes + // it: the new primary's numbers are at least as fresh as any probe + // answer. + self.finish_view_probe(); self.view.set(msg_view); self.log_view.set(msg_view); self.status.set(Status::Normal); self.ceded_primaryship.set(false); self.advance_commit_max(msg_commit); - self.fast_forward_recovering_commit_floor(); self.reset_view_change_state(); // Stale pipeline entries from the old view must be discarded @@ -1947,13 +2134,6 @@ impl> VsrConsensus { let old_commit_max = self.commit_max.get(); self.advance_commit_max(header.commit); - if self.recovering.get() { - // First learned commit point after a state-less boot: the - // recovered durable data stands in for the journal prefix, so - // there is nothing local to apply for it. - self.fast_forward_recovering_commit_floor(); - return CommitOutcome::Accepted; - } if self.commit_max.get() > old_commit_max { CommitOutcome::Advanced } else { @@ -1961,25 +2141,6 @@ impl> VsrConsensus { } } - /// See the `recovering` field: align `commit_min` with the learned - /// `commit_max` exactly once, then leave recovery mode. - fn fast_forward_recovering_commit_floor(&self) { - if !self.recovering.get() { - return; - } - self.recovering.set(false); - let commit_max = self.commit_max.get(); - if commit_max > self.commit_min.get() { - tracing::info!( - replica = self.replica, - namespace_raw = self.namespace, - commit_max, - "recovering replica adopting cluster commit floor" - ); - self.commit_min.set(commit_max); - } - } - /// Complete view change as the new primary after collecting DVC quorum. /// /// # Client-table maintenance @@ -2031,6 +2192,7 @@ impl> VsrConsensus { timeouts.stop(TimeoutKind::ViewChangeStatus); timeouts.stop(TimeoutKind::DoViewChangeMessage); timeouts.stop(TimeoutKind::StartViewChangeMessage); + timeouts.stop(TimeoutKind::RequestStartViewMessage); timeouts.start(TimeoutKind::CommitMessage); // If there are uncommitted ops in the rebuilt pipeline, start the // Prepare timeout so that lost PrepareOks trigger retransmission. diff --git a/core/consensus/src/observability.rs b/core/consensus/src/observability.rs index 91d85bb207..03a9b1b088 100644 --- a/core/consensus/src/observability.rs +++ b/core/consensus/src/observability.rs @@ -56,6 +56,15 @@ impl ReplicaRole { #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ViewChangeReason { NormalHeartbeatTimeout, + /// The current view's primary sent a `RequestStartView` probe: it + /// cannot lead (restarted, or abandoned the view mid-view-change), so + /// peers elect without waiting for its heartbeats to stop arriving + /// (they already have). + PrimaryProbedView, + /// A recovering replica's `RequestStartView` probes all went unanswered: + /// nobody in the cluster is settled (full-cluster restart), so waiting + /// for a primary is futile -- elect instead. + ViewProbeUnanswered, ViewChangeStatusTimeout, ReceivedStartViewChange, ReceivedDoViewChange, @@ -66,6 +75,8 @@ impl ViewChangeReason { pub const fn as_str(self) -> &'static str { match self { Self::NormalHeartbeatTimeout => "normal_heartbeat_timeout", + Self::PrimaryProbedView => "primary_probed_view", + Self::ViewProbeUnanswered => "view_probe_unanswered", Self::ViewChangeStatusTimeout => "view_change_status_timeout", Self::ReceivedStartViewChange => "received_start_view_change", Self::ReceivedDoViewChange => "received_do_view_change", @@ -111,6 +122,7 @@ impl IgnoreReason { #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ControlActionKind { + SendRequestStartView, SendStartViewChange, SendDoViewChange, SendStartView, @@ -125,6 +137,7 @@ impl ControlActionKind { #[must_use] pub const fn as_str(self) -> &'static str { match self { + Self::SendRequestStartView => "send_request_start_view", Self::SendStartViewChange => "send_start_view_change", Self::SendDoViewChange => "send_do_view_change", Self::SendStartView => "send_start_view", @@ -308,6 +321,13 @@ impl ControlActionLogEvent { #[must_use] pub const fn from_vsr_action(replica: ReplicaLogContext, action: &VsrAction) -> Self { match *action { + VsrAction::SendRequestStartView { .. } => Self { + replica, + action: ControlActionKind::SendRequestStartView, + target_replica: None, + op: None, + commit: None, + }, VsrAction::SendStartViewChange { .. } => Self { replica, action: ControlActionKind::SendStartViewChange, diff --git a/core/consensus/src/vsr_timeout.rs b/core/consensus/src/vsr_timeout.rs index eabc477e6d..eb7076a116 100644 --- a/core/consensus/src/vsr_timeout.rs +++ b/core/consensus/src/vsr_timeout.rs @@ -106,6 +106,8 @@ pub enum TimeoutKind { StartViewChangeMessage, ViewChangeStatus, DoViewChangeMessage, + /// Backup re-requesting the current view's `StartView` from its + /// primary: fired while a recovering replica probes for the view. RequestStartViewMessage, } @@ -126,7 +128,7 @@ pub struct TimeoutManager { #[allow(unused)] impl TimeoutManager { - // Timeout durations in ticks (10ms per tick). Values are taken from TB. + // Timeout durations in ticks (10ms per tick). // TODO define 10ms per tick in a separate constant. const PING_TICKS: u64 = 100; const PREPARE_TICKS: u64 = 25; diff --git a/core/integration/src/harness/orchestrator/harness.rs b/core/integration/src/harness/orchestrator/harness.rs index 31320235d8..df1d545e57 100644 --- a/core/integration/src/harness/orchestrator/harness.rs +++ b/core/integration/src/harness/orchestrator/harness.rs @@ -330,6 +330,36 @@ impl TestHarness { Ok(()) } + /// Restart EVERY node and reconnect all clients: the full-cluster + /// restart path, where no settled primary survives to answer the rejoin + /// probes and the replicas must fall back to an election among their + /// recovered logs. All nodes stop BEFORE any starts, so no rejoiner can + /// lean on a still-live peer (that would be a rolling restart). + pub async fn restart_cluster(&mut self) -> Result<(), TestBinaryError> { + if self.servers.is_empty() { + return Err(TestBinaryError::MissingServer); + } + + for client in &mut self.clients { + client.disconnect().await; + } + + for server in self.servers.iter_mut().rev() { + server.stop_dependents()?; + server.stop()?; + } + for server in &mut self.servers { + server.restart()?; + } + + self.update_client_addresses(); + for client in &mut self.clients { + client.connect().await?; + } + + Ok(()) + } + /// Get reference to the first (primary) server handle. pub fn server(&self) -> &ServerHandle { self.servers.first().expect("No servers configured") diff --git a/core/integration/tests/data_integrity/mod.rs b/core/integration/tests/data_integrity/mod.rs index 7ec214d7b3..8bb56bb0d3 100644 --- a/core/integration/tests/data_integrity/mod.rs +++ b/core/integration/tests/data_integrity/mod.rs @@ -15,9 +15,12 @@ // specific language governing permissions and limitations // under the License. -// Restart-based verifiers stay `not(vsr)`: they restart cluster nodes and -// assert on-disk state afterward, which VSR can't satisfy until replica state -// transfer lands (a restarted replica has no way to catch up). +// Requires state transfer: the 5 MB bench fill is thousands of ops while the +// partition journal's evicted ring retains only the last 4096, so a restarted +// replica's rejoin window exceeds what journal repair can serve. The commit +// floor lets recovered segments stand in for the evicted prefix, but the +// sub-floor stats/offset seeding this test asserts (exact messages_count / +// size_bytes across the restart) is state transfer's job. #[cfg(not(feature = "vsr"))] mod verify_after_server_restart; mod verify_user_login_after_restart; @@ -28,8 +31,7 @@ mod verify_user_login_after_restart; mod verify_no_plaintext_credentials_on_disk; // The cooperative-rebalance matrix runs under vsr too: it exercises server-ng's -// consumer-group rebalancing (a VSR feature) and never restarts the cluster, so -// the no-state-transfer limitation above does not apply. Green at 95/95. +// consumer-group rebalancing (a VSR feature). Green at 95/95. mod verify_consumer_group_partition_assignment; // Cross-replica on-disk data identity is VSR-only. diff --git a/core/integration/tests/mod.rs b/core/integration/tests/mod.rs index c4171549a4..84b46be4e4 100644 --- a/core/integration/tests/mod.rs +++ b/core/integration/tests/mod.rs @@ -30,8 +30,10 @@ use tracing_subscriber::layer::SubscriberExt; use tracing_subscriber::util::SubscriberInitExt; use tracing_subscriber::{EnvFilter, fmt}; -// Drives the `iggy` CLI binary against a running server. Untriaged for vsr: -// nobody has assessed which of its ~170 cases work against server-ng. +// Drives the `iggy` CLI binary against a running server. Probed against +// server-ng 2026-07-15: ~109/170 pass; the ~42 failures + hangs cluster in +// `context` (12), `system` (7), `message` (6), `stream` (5) plus scattered +// others -- needs a dedicated triage pass before un-gating. #[cfg(not(feature = "vsr"))] mod cli; // A single `#[ignore]`d multi-node ping matrix stub; none of its cells run in @@ -40,8 +42,8 @@ mod cli; mod cluster; mod config_provider; mod connectors; -// Runs under vsr. The remaining per-module gap (`verify_after_server_restart`) -// is gated inside the module with the reason (replica state transfer). +// Runs under vsr; the one gap (`verify_after_server_restart`) is gated inside +// the module: its rejoin window exceeds journal retention (state transfer). mod data_integrity; mod mcp; mod sdk; diff --git a/core/integration/tests/sdk/mcp_parity.rs b/core/integration/tests/sdk/mcp_parity.rs index cd27ba25aa..4ec5422af0 100644 --- a/core/integration/tests/sdk/mcp_parity.rs +++ b/core/integration/tests/sdk/mcp_parity.rs @@ -329,12 +329,7 @@ async fn should_return_clients(harness: &TestHarness) { } // mcp::should_handle_snapshot -// The diagnostic `snapshot` op (heap/config/process dump) is not implemented in -// server-ng; it falls through to the non-replicated catch-all and returns an -// empty body. Out of scope for metadata/stats parity -- un-ignore if server-ng -// grows the snapshot subsystem. #[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)] -#[ignore = "vsr: server-ng does not implement the diagnostic snapshot op"] async fn should_handle_snapshot(harness: &TestHarness) { let client = root_client(harness).await; let snapshot = client diff --git a/core/integration/tests/server/mod.rs b/core/integration/tests/server/mod.rs index 4e88fb9692..015df6978d 100644 --- a/core/integration/tests/server/mod.rs +++ b/core/integration/tests/server/mod.rs @@ -49,9 +49,8 @@ mod general; mod message_cleanup; mod message_retrieval; // Server restarts, consumer-group barriers, and DeleteSegments maintenance. -// `should_delete_segments_without_consumers` runs fully (both restart -// variants) under server-ng; the consumer-variant restart cells are vsr-gated -// on replica state transfer (reasons on the gates inside). +// The full restart matrix (consumer variants included) runs under server-ng: +// a restarted replica rejoins via the view probe + journal repair. mod purge_delete; mod scenarios; mod specific; diff --git a/core/integration/tests/server/purge_delete.rs b/core/integration/tests/server/purge_delete.rs index ccf0a1679f..5b631ae608 100644 --- a/core/integration/tests/server/purge_delete.rs +++ b/core/integration/tests/server/purge_delete.rs @@ -25,16 +25,7 @@ use test_case::test_matrix; partition.messages_required_to_save = "1", partition.enforce_fsync = "true", ))] -// The consumer polls in this scenario auto-commit offsets, so partition ops -// keep flowing while the restarted node rejoins. Ops prepared in that window -// commit on the surviving quorum and leave the pipeline before the rejoining -// replica can ack them, so its journal has a gap it cannot fill without -// message repair; when the commit frontier crosses the gap the replica -// correctly suicides ("replica is divergent"). Re-enable restart_on once -// message repair lands. The consumerless variant below exercises the -// restart + prune path without this window. -#[cfg_attr(not(feature = "vsr"), test_matrix([restart_off(), restart_on()]))] -#[cfg_attr(feature = "vsr", test_matrix([restart_off()]))] +#[test_matrix([restart_off(), restart_on()])] async fn should_delete_segments_and_validate_filesystem( harness: &mut TestHarness, restart_server: bool, @@ -72,14 +63,7 @@ async fn should_delete_segments_with_consumer_group_barrier(harness: &TestHarnes partition.messages_required_to_save = "1", partition.enforce_fsync = "true", ))] -// Consumer polls auto-commit offsets, so partition ops keep flowing while the -// restarted node rejoins. Ops committed by the surviving quorum in that window -// leave the pipeline before the rejoining replica can receive them; without -// state transfer its journal gap is unfillable and the replica correctly -// suicides ("replica is divergent") when the commit frontier crosses it. -// Re-enable restart_on once state transfer lands. -#[cfg_attr(not(feature = "vsr"), test_matrix([restart_off(), restart_on()]))] -#[cfg_attr(feature = "vsr", test_matrix([restart_off()]))] +#[test_matrix([restart_off(), restart_on()])] async fn should_block_deletion_until_all_consumers_pass_segment( harness: &mut TestHarness, restart_server: bool, @@ -95,13 +79,8 @@ async fn should_block_deletion_until_all_consumers_pass_segment( ))] // The scenario asserts the exact [0, 7, 14, 21] layout only on the legacy path; // under vsr it verifies the framing-agnostic purge outcome (offsets cleared, -// files deleted, partition reset to a single segment at offset 0). restart_on -// is vsr-gated on the same rejoin-window state-transfer gap as -// `should_block_deletion_until_all_consumers_pass_segment` above (the old -// CG-reconnect `AlreadyAuthenticated` blocker is fixed by the SDK's -// fresh-session reconnect). -#[cfg_attr(not(feature = "vsr"), test_matrix([restart_off(), restart_on()]))] -#[cfg_attr(feature = "vsr", test_matrix([restart_off()]))] +// files deleted, partition reset to a single segment at offset 0). +#[test_matrix([restart_off(), restart_on()])] async fn should_purge_topic_and_clear_consumer_offsets( harness: &mut TestHarness, restart_server: bool, diff --git a/core/integration/tests/server/scenarios/mod.rs b/core/integration/tests/server/scenarios/mod.rs index d69f61c902..5d701d15e5 100644 --- a/core/integration/tests/server/scenarios/mod.rs +++ b/core/integration/tests/server/scenarios/mod.rs @@ -36,10 +36,6 @@ pub mod create_message_payload; pub mod cross_protocol_pat_scenario; pub mod encryption_scenario; pub mod invalid_consumer_offset_scenario; -// Asserts server log-file rotation/archival policies; server-ng's file -// logger only captures bootstrap output (shard-thread logs never reach the -// file), so volume-based rotation rules cannot trigger. -#[cfg(not(feature = "vsr"))] pub mod log_rotation_scenario; pub mod message_cleanup_scenario; pub mod message_headers_scenario; diff --git a/core/integration/tests/server/scenarios/purge_delete_scenario.rs b/core/integration/tests/server/scenarios/purge_delete_scenario.rs index a3d0fc6fdc..ee1bf2ac46 100644 --- a/core/integration/tests/server/scenarios/purge_delete_scenario.rs +++ b/core/integration/tests/server/scenarios/purge_delete_scenario.rs @@ -145,11 +145,14 @@ pub async fn run(harness: &mut TestHarness, restart_server: bool) { await_segment_layout(&partition_path, &EXPECTED_SEGMENT_OFFSETS[1..]).await; assert_segment_file_sizes(&partition_path, &EXPECTED_SEGMENT_OFFSETS[1..]); - assert_eq!( - poll_all_offsets(&client, &stream_ident, &topic_ident).await, + await_polled_offsets( + &client, + &stream_ident, + &topic_ident, (MSGS_PER_SEALED_SEGMENT..TOTAL_MESSAGES as u64).collect::>(), - "Messages in the remaining segments survive" - ); + "Messages in the remaining segments survive", + ) + .await; // After deleting segment 0 (7 messages removed): current_offset must still // reflect the true partition max (24), not messages_count - 1 (17). @@ -201,11 +204,14 @@ pub async fn run(harness: &mut TestHarness, restart_server: bool) { await_segment_layout(&partition_path, &EXPECTED_SEGMENT_OFFSETS[2..]).await; assert_segment_file_sizes(&partition_path, &EXPECTED_SEGMENT_OFFSETS[2..]); - assert_eq!( - poll_all_offsets(&client, &stream_ident, &topic_ident).await, + await_polled_offsets( + &client, + &stream_ident, + &topic_ident, (2 * MSGS_PER_SEALED_SEGMENT..TOTAL_MESSAGES as u64).collect::>(), - "Messages 14..25 survive" - ); + "Messages 14..25 survive", + ) + .await; // After deleting segments 0 and 1 (14 messages removed): current_offset // must still be 24, not messages_count - 1 (10). @@ -282,21 +288,16 @@ pub async fn run(harness: &mut TestHarness, restart_server: bool) { &partition_path, std::slice::from_ref(&active_segment_offset), ); - assert_eq!( - poll_all_offsets(&client, &stream_ident, &topic_ident).await, + await_polled_offsets( + &client, + &stream_ident, + &topic_ident, (active_segment_offset..TOTAL_MESSAGES as u64).collect::>(), - "Messages in the active segment survive" - ); + "Messages in the active segment survive", + ) + .await; - // --- Error cases --- - // - // Legacy rejects a delete on an unknown target. server-ng acks it without - // committing anything (best-effort space management), which gaps the VSR - // request sequence: the next metadata op hits the `request == committed + 1` - // preflight, gets dropped, and the SDK read-timeout replay panics - // (`register_request_id already called`). Skipped under vsr until - // unresolvable targets are rejected (or committed as no-ops) at admission. - #[cfg(not(feature = "vsr"))] + // --- Error cases: deletes on unknown targets must be rejected --- { assert!( client diff --git a/core/integration/tests/server/scenarios/reconnect_after_restart_scenario.rs b/core/integration/tests/server/scenarios/reconnect_after_restart_scenario.rs index b22bf153ea..e994e5d443 100644 --- a/core/integration/tests/server/scenarios/reconnect_after_restart_scenario.rs +++ b/core/integration/tests/server/scenarios/reconnect_after_restart_scenario.rs @@ -15,7 +15,6 @@ // specific language governing permissions and limitations // under the License. -#[cfg(not(feature = "vsr"))] use futures::StreamExt; use iggy::prelude::*; use iggy_common::TransportProtocol; @@ -100,7 +99,6 @@ pub async fn run_producer(harness: &mut TestHarness) { ); } -#[cfg(not(feature = "vsr"))] pub async fn run_consumer(harness: &mut TestHarness) { let setup_client = harness .root_client() @@ -211,7 +209,6 @@ fn create_client(harness: &TestHarness) -> IggyClient { .expect("Failed to create client from connection string") } -#[cfg(not(feature = "vsr"))] async fn send_messages(client: &IggyClient, prefix: &str, count: u32) { for i in 0..count { let msg = IggyMessage::from_str(&format!("{prefix}-{i}")).unwrap(); @@ -227,7 +224,6 @@ async fn send_messages(client: &IggyClient, prefix: &str, count: u32) { } } -#[cfg(not(feature = "vsr"))] /// Consumes up to `expected` messages, returning their payloads in order. async fn consume_messages_validated( consumer: &mut IggyConsumer, @@ -511,3 +507,208 @@ pub async fn run_consumer_offset_ahead_after_crash(harness: &mut TestHarness) { "BUG: consumer skipped to offset {first_offset}, expected messages in range 10-14" ); } + +/// Full-cluster restart: every node stops before any comes back, so no +/// settled primary survives to answer the rejoin probes. The replicas must +/// give up probing (`ViewChangeReason::ViewProbeUnanswered`), elect among +/// their recovered logs, and serve the durable data across the outage. The +/// caller's server config must flush eagerly (`messages_required_to_save=1` +/// + fsync): with every journal dying at once, only flushed bytes survive. +pub async fn run_full_cluster_restart(harness: &mut TestHarness) { + 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, "pre-outage", 10).await; + drop(setup_client); + // Let the backups' commit walk flush the tail: the send ack proves quorum + // COMMIT, but each replica flushes on its own walk (driven by the commit + // heartbeat), and the partition journal does not survive the process. A + // kill racing the last heartbeat leaves the newest message durable only + // on the old primary -- the known journal-durability caveat, not what + // this scenario tests. + sleep(Duration::from_secs(2)).await; + + harness + .restart_cluster() + .await + .expect("Failed to restart the whole cluster"); + + let client = harness + .root_client() + .await + .expect("Failed to reconnect after full-cluster restart"); + let survivors = poll_from_zero_until(&client, 10, Duration::from_secs(30)).await; + assert_eq!( + survivors, + (0..10) + .map(|i| format!("pre-outage-{i}")) + .collect::>(), + "flushed messages must survive a full-cluster restart" + ); + + send_messages(&client, "post-outage", 10).await; + let all = poll_from_zero_until(&client, 20, Duration::from_secs(30)).await; + assert_eq!(all.len(), 20, "the reformed cluster must accept new writes"); + assert_eq!(all[10], "post-outage-0"); + assert_eq!(all[19], "post-outage-9"); +} + +/// Polls from offset 0 until `expected` messages arrive or the deadline +/// passes, returning their payloads. The cluster may still be mid-election +/// right after a restart, so a single poll is not a fair snapshot. +async fn poll_from_zero_until( + client: &IggyClient, + expected: u32, + deadline: Duration, +) -> Vec { + let started = std::time::Instant::now(); + loop { + let polled = client + .poll_messages( + &Identifier::named(STREAM_NAME).unwrap(), + &Identifier::named(TOPIC_NAME).unwrap(), + Some(0), + &Consumer::default(), + &PollingStrategy::offset(0), + expected * 2, + false, + ) + .await; + if let Ok(polled) = &polled + && polled.messages.len() >= expected as usize + { + return polled + .messages + .iter() + .map(|message| String::from_utf8_lossy(&message.payload).to_string()) + .collect(); + } + if started.elapsed() > deadline { + let got = polled.map(|p| p.messages.len()).unwrap_or(0); + panic!("expected {expected} messages after full-cluster restart, got {got}"); + } + 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() +} diff --git a/core/integration/tests/server/specific.rs b/core/integration/tests/server/specific.rs index 479d502b0b..0c455d045e 100644 --- a/core/integration/tests/server/specific.rs +++ b/core/integration/tests/server/specific.rs @@ -72,18 +72,12 @@ async fn producer_reconnect_after_server_restart(harness: &mut TestHarness) { reconnect_after_restart_scenario::run_producer(harness).await; } -// vsr-gated on the rejoin-window state-transfer gap: the consumer's polls -// auto-commit offsets, so an offset op in flight at the kill commits on the -// surviving quorum and the restarted replica can never fetch it -- when the -// commit frontier crosses the gap it correctly suicides ("replica is -// divergent"). Racy (the window is only sometimes non-empty), so it flakes -// rather than fails deterministically. The producer variant stays un-gated: -// its sends are acked before the kill, leaving an empty window. QUIC has an -// additional SDK gap (post-reconnect consumer polls return nothing; no -// mid-connection failover), so it stays gated even once state transfer lands -// unless that is fixed first. -#[cfg(not(feature = "vsr"))] -#[iggy_harness( +// QUIC stays vsr-gated on an SDK gap: after the restart the QUIC client +// redirects to the new leader, reconnects, and signs in, but the long-lived +// consumer's polls then return nothing for the whole window -- the +// post-reconnect request path wedges (QUIC also lacks the TCP client's +// mid-connection failover). TCP and WebSocket run. +#[cfg_attr(not(feature = "vsr"), iggy_harness( test_client_transport = [Tcp, WebSocket, Quic], server( tcp.socket.override_defaults = true, @@ -91,7 +85,16 @@ async fn producer_reconnect_after_server_restart(harness: &mut TestHarness) { quic.max_idle_timeout = "500s", quic.keep_alive_interval = "15s" ) -)] +))] +#[cfg_attr(feature = "vsr", iggy_harness( + test_client_transport = [Tcp, WebSocket], + server( + tcp.socket.override_defaults = true, + tcp.socket.nodelay = true, + quic.max_idle_timeout = "500s", + quic.keep_alive_interval = "15s" + ) +))] async fn consumer_reconnect_after_server_restart(harness: &mut TestHarness) { reconnect_after_restart_scenario::run_consumer(harness).await; } @@ -104,6 +107,30 @@ async fn single_message_restart_offset_zero(harness: &mut TestHarness) { reconnect_after_restart_scenario::run_single_message_offset_zero_restart(harness).await; } +// Full-cluster restart is vsr-only by construction: it exercises the rejoin +// probe's election fallback across all replicas, which a single-process +// legacy server has no equivalent of (plain restart covers it there). +#[cfg(feature = "vsr")] +#[iggy_harness(server( + partition.messages_required_to_save = "1", + partition.enforce_fsync = true +))] +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 diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 7e7efec7fb..53b1bb3f13 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -26,6 +26,7 @@ use crate::poll_plan::{ ResidentTailSnapshot, }; use crate::segment::Segment; +use crate::types::RepairSession; use crate::{ AppendResult, Partition, PartitionOffsets, PartitionsConfig, PollQueryResult, PollingArgs, PollingConsumer, @@ -42,7 +43,9 @@ use iggy_binary_protocol::requests::consumer_offsets::{ DeleteConsumerOffset2Request, DeleteConsumerOffsetRequest, StoreConsumerOffset2Request, StoreConsumerOffsetRequest, }; -use iggy_binary_protocol::{AckLevel, Operation, PrepareHeader, WireDecode, WireIdentifier}; +use iggy_binary_protocol::{ + AckLevel, GenericHeader, Operation, PrepareHeader, WireDecode, WireIdentifier, +}; use iggy_binary_protocol::{PrepareOkHeader, RequestHeader}; use iggy_common::{ ConsumerGroupId, ConsumerGroupOffsets, ConsumerKind, ConsumerOffset, ConsumerOffsets, @@ -51,8 +54,8 @@ use iggy_common::{ use journal::Journal as _; use message_bus::{IggyMessageBus, MessageBus, is_auto_commit_client}; use server_common::{ - Message, SegmentStorage, - iobuf::Frozen, + MESSAGE_ALIGN, Message, SegmentStorage, + iobuf::{Frozen, Owned}, send_messages2::{ convert_request_message, decode_prepare_slice, stamp_prepare_for_persistence, }, @@ -106,6 +109,16 @@ where /// `None` only for in-memory (simulated) partitions. partition_dir: Option, consumer_offset_enforce_fsync: bool, + /// In-flight journal repair: + /// set when the recovery handshake finds this replica behind the group's + /// commit frontier, cleared when `RepairDone` completes the walk. + pub repair: Option, + /// Highest message offset recovered from segments at boot (`None` when + /// the partition booted empty). Repaired batches at or below this line + /// are already persisted and counted; the flush and commit paths skip + /// re-persisting / re-counting them. Immutable after boot, so live + /// traffic (always above it) is never affected. + pub recovered_durable_offset: Option, pending_consumer_offset_commits: HashMap, /// Committed-only mirror of each consumer's persisted offset file: the /// last value this replica durably wrote per (kind, consumer id). Fed @@ -212,7 +225,8 @@ where { pub fn new(stats: Arc, consensus: VsrConsensus) -> Self { let observed_view = consensus.view(); - Self { + let single_replica = consensus.replica_count() == 1; + let partition = Self { consensus, log: SegmentedLog::default(), offset: Arc::new(AtomicU64::new(0)), @@ -229,11 +243,17 @@ where consumer_group_offsets_path: None, partition_dir: None, consumer_offset_enforce_fsync: false, + repair: None, + recovered_durable_offset: None, pending_consumer_offset_commits: HashMap::new(), persisted_offsets: RefCell::new(HashMap::new()), observed_view, applied_purge_generation: 0, + }; + if single_replica { + partition.log.journal().inner.set_repair_retention(false); } + partition } #[must_use] @@ -339,10 +359,16 @@ where // fails (e.g. disk full, fd exhausted) the pending entry must remain // stageable for retry on the next apply. Removing first would strand // the op - not on disk AND not in memory. - let pending = *self - .pending_consumer_offset_commits - .get(&op) - .ok_or(IggyError::InvalidCommand)?; + let pending = match self.pending_consumer_offset_commits.get(&op) { + Some(pending) => *pending, + // A view change clears the staged table (uncommitted ops may be + // superseded by the new view's log), and suffixes adopted via + // DoViewChange/StartView or journal repair never pass the live + // staging path at all. The journal entry IS the new view's + // authoritative content for this op, so re-derive the commit + // from it instead of wedging the commit walk. + None => self.restage_consumer_offset_from_journal(op)?, + }; // Persist to the on-disk offset table first so a crash after the // in-memory apply cannot observe a readable offset that was not // durably stored; the in-memory update is idempotent on replay @@ -1876,6 +1902,17 @@ where if message_count == 0 { continue; } + // A repaired batch at or below the boot-time recovered + // durable offset is already IN the segments this replica + // recovered; persisting it again would append duplicate + // bytes past the segment end. Evict it without writing. + // Live traffic always sits above the (immutable) line. + let batch_end = batch.header.base_offset + u64::from(message_count) - 1; + if let Some(durable) = self.recovered_durable_offset + && batch_end <= durable + { + continue; + } if flush_index.is_none() { // Record only; the in-mem cache insert is deferred until the @@ -2202,15 +2239,24 @@ where } if let Some(batch_stats) = committed_visible_offsets.get(&prepare_header.op) { - self.offset.store(batch_stats.end_offset, Ordering::Release); - self.stats.set_current_offset(batch_stats.end_offset); - // Advance the aggregate stats with the visible offset. Disk - // persistence is threshold-gated in `commit_messages`, which - // must not also touch these counters or committed messages - // would be double-counted once they flush. - self.stats - .increment_messages_count(u64::from(batch_stats.message_count)); - self.stats.increment_size_bytes(batch_stats.size_bytes); + // A repaired batch at or below the boot-time recovered + // durable offset was already counted (and persisted) + // before the restart; skip it. Live traffic always sits + // above the (immutable) line. + if self + .recovered_durable_offset + .is_none_or(|durable| batch_stats.end_offset > durable) + { + self.offset.store(batch_stats.end_offset, Ordering::Release); + self.stats.set_current_offset(batch_stats.end_offset); + // Advance the aggregate stats with the visible offset. Disk + // persistence is threshold-gated in `commit_messages`, which + // must not also touch these counters or committed messages + // would be double-counted once they flush. + self.stats + .increment_messages_count(u64::from(batch_stats.message_count)); + self.stats.increment_size_bytes(batch_stats.size_bytes); + } } !*failed_commit } @@ -2298,6 +2344,43 @@ where } } + fn restage_consumer_offset_from_journal( + &self, + op: u64, + ) -> Result { + let entry = self + .log + .journal() + .inner + .repair_entry(op) + .ok_or(IggyError::InvalidCommand)?; + // Deep copy: the journal buffer is shared and `Message::try_from` + // wants an `Owned`; this path only runs on the post-view-change + // fallback, never per-commit. + let owned = Owned::::copy_from_slice(entry.as_slice()); + let message = Message::::try_from(owned) + .map_err(|_| IggyError::InvalidCommand)? + .try_into_typed::() + .map_err(|_| IggyError::InvalidCommand)?; + let header = *message.header(); + let (kind, consumer_id, offset, _ack) = + Self::parse_staged_consumer_offset_commit(header.operation, &message)?; + match header.operation { + Operation::StoreConsumerOffset | Operation::StoreConsumerOffset2 => { + let offset = offset.ok_or(IggyError::InvalidCommand)?; + Ok(if is_auto_commit_client(header.client) { + PendingConsumerOffsetCommit::upsert_auto_commit(kind, consumer_id, offset) + } else { + PendingConsumerOffsetCommit::upsert(kind, consumer_id, offset) + }) + } + Operation::DeleteConsumerOffset | Operation::DeleteConsumerOffset2 => { + Ok(PendingConsumerOffsetCommit::delete(kind, consumer_id)) + } + _ => Err(IggyError::InvalidCommand), + } + } + fn parse_staged_consumer_offset_commit( operation: Operation, message: &Message, @@ -2936,6 +3019,211 @@ where nth_oldest_sealed_end(self.log.segments(), count) } + /// Ingest one repaired prepare: journal + stage it exactly like a live + /// replicated op, minus the view fence, the gap check, and the ack (the + /// op is already committed cluster-wide; there is nobody to ack to). The + /// commit walk runs at `RepairDone`, after the floor is known. + pub async fn apply_repaired_prepare(&mut self, message: Message) { + let header = *message.header(); + let Some(session) = &self.repair else { + return; + }; + if header.op <= self.consensus().commit_min() || header.op > session.to_op { + return; + } + // Any in-window frame proves the stream is alive; only silence + // should age the stall counter. + if let Some(session) = self.repair.as_mut() { + session.idle_ticks = 0; + } + if self.log.journal().inner.header_by_op(header.op).is_some() { + return; + } + let applied = if header.operation == Operation::SendMessages { + match self.append_repaired_send_messages(message).await { + Ok(base_offset) => { + if let (Some(base_offset), Some(session)) = (base_offset, self.repair.as_mut()) + { + session.first_batch_offset = Some( + session + .first_batch_offset + .map_or(base_offset, |first| first.min(base_offset)), + ); + } + Ok(()) + } + Err(error) => Err(error), + } + } else { + self.apply_replicated_operation(message).await + }; + if let Err(error) = applied { + warn!( + target: "iggy.partitions.diag", + plane = "partitions", + namespace_raw = self.namespace().inner(), + op = header.op, + %error, + "failed to journal repaired prepare" + ); + return; + } + // Advance the sequencer only along the CONTIGUOUS journaled + // frontier. DVC advertises `op = sequencer.current_sequence()` and + // elections pick the max, so bumping straight to a repaired op that + // sits above an unfilled hole would let this replica win a view it + // cannot walk. A dropped frame stalls the frontier here; the stall + // retry refills the hole and the next apply resumes the advance + // (walking over ops that were journaled out of order meanwhile). + let mut frontier = self.consensus().sequencer().current_sequence(); + while self + .log + .journal() + .inner + .header_by_op(frontier + 1) + .is_some() + { + frontier += 1; + } + let consensus = self.consensus(); + if frontier > consensus.sequencer().current_sequence() { + consensus.sequencer().set_sequence(frontier); + } + consensus.set_last_prepare_checksum(header.checksum); + } + + /// Conclude a repair stream: settle the commit floor at the serving + /// peer's eviction point (everything below it is represented by this + /// replica's recovered segments + offset files) and walk the repaired + /// window through the normal commit path. + pub async fn complete_repair(&mut self, config: &PartitionsConfig) { + let Some(session) = self.repair else { + return; + }; + if let Some(floor) = session.floor { + // A peer may have evicted past this replica's commit frontier; + // an unclamped floor would drive commit_min above commit_max and + // panic the next advance. + let floor = floor.min(self.consensus().commit_max()); + // The floor claims "recovered durable state stands in below me". + // Verify it: the served window must connect to the recovered + // segments. A window starting above the durable end means ops + // below the floor are neither locally durable nor repaired -- + // that gap is state-transfer territory, and accepting the floor + // would silently serve a holed log. Refuse and stay gap-stopped: + // a visible stall beats invisible loss. + let durable_end = self.recovered_durable_offset; + let connected = match (session.first_batch_offset, durable_end) { + (Some(first), Some(durable)) => first <= durable.saturating_add(1), + (Some(first), None) => first == 0, + // Offsets-only window: nothing to reconcile against segments; + // the consumer-offset table on disk stands in below the floor. + (None, _) => true, + }; + if !connected { + tracing::error!( + target: "iggy.partitions.diag", + plane = "partitions", + namespace_raw = self.namespace().inner(), + floor, + first_batch_offset = ?session.first_batch_offset, + recovered_durable_offset = ?durable_end, + "refusing commit floor: repaired window does not connect \ + to recovered durable state (needs state transfer)" + ); + self.commit_journal(config).await; + return; + } + let commit_min = self.consensus().commit_min(); + if floor > commit_min { + self.consensus().set_commit_floor(floor); + } + } + let before = self.consensus().commit_min(); + self.commit_journal(config).await; + let commit_min = self.consensus().commit_min(); + // Completion is decided HERE, not by the peer's served-through + // claim: repair frames ride a lossy best-effort bus, so a stream + // the peer fully served can still arrive with holes. Only a walk + // that reached the requested frontier closes the session; anything + // less keeps it armed and the stall retry re-requests the remains + // (`commit_min + 1..`), converging over rounds. + let done = commit_min >= session.to_op; + if done { + self.repair = None; + } + tracing::info!( + target: "iggy.partitions.diag", + plane = "partitions", + namespace_raw = self.namespace().inner(), + commit_min_before = before, + commit_min_after = commit_min, + commit_max = self.consensus().commit_max(), + to_op = session.to_op, + done, + "repair window commit walk finished" + ); + } + + /// Journal a repaired `SendMessages` prepare, preserving its embedded + /// batch stamps. A stored prepare was stamped by `append_messages` on + /// the serving replica BEFORE it was journaled, so its `base_offset` / + /// `base_timestamp` / `batch_checksum` are the canonical values every + /// replica agreed on. Re-stamping from this replica's dirty counter + /// (what the live path does) mints a second copy of the window at + /// fresh offsets whenever recovered segments already hold the + /// originals: the counter sits at the recovered durable END, not at + /// the op's position in history. + async fn append_repaired_send_messages( + &mut self, + message: Message, + ) -> Result, IggyError> { + let write_lock = self.write_lock.clone(); + let _guard = write_lock.lock().await; + + let (base_offset, base_timestamp, total_size, message_count) = { + let batch = + decode_prepare_slice(message.as_slice()).map_err(|_| IggyError::InvalidCommand)?; + ( + batch.header.base_offset, + batch.header.base_timestamp, + batch.header.total_size() as u64, + batch.message_count(), + ) + }; + if message_count == 0 { + return Ok(None); + } + let last_offset = base_offset + u64::from(message_count) - 1; + + self.should_increment_offset = true; + let dirty = self.dirty_offset.load(Ordering::Relaxed); + self.dirty_offset + .store(dirty.max(last_offset), Ordering::Relaxed); + + let segment_index = self.log.segments().len() - 1; + let current_position = self.log.segments()[segment_index].current_position; + self.log.segments_mut()[segment_index].current_position = current_position + .checked_add(total_size) + .ok_or(IggyError::CannotAppendMessage)?; + + let journal = self.log.journal_mut(); + journal.info.messages_count += message_count; + journal.info.size += IggyByteSize::from(total_size); + journal.info.current_offset = last_offset; + if journal.info.first_timestamp == 0 { + journal.info.first_timestamp = base_timestamp; + } + journal.info.end_timestamp = base_timestamp; + journal.info.max_timestamp = journal.info.max_timestamp.max(base_timestamp); + journal + .inner + .append(message.into_frozen()) + .await + .map_err(|_| IggyError::CannotAppendMessage)?; + Ok(Some(base_offset)) + } + async fn send_prepare_ok(&self, header: &PrepareHeader) { // `VsrAction::RetransmitPrepares` reads from `self.log.journal`. // Both `SendMessages` (via `append_send_messages_to_journal`) and diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs index 131831aaaf..feb78c24b1 100644 --- a/core/partitions/src/journal.rs +++ b/core/partitions/src/journal.rs @@ -23,8 +23,8 @@ use server_common::{ }; use std::io; use std::{ - cell::UnsafeCell, - collections::{BTreeMap, HashMap}, + cell::{Cell, UnsafeCell}, + collections::{BTreeMap, HashMap, VecDeque}, }; use tracing::warn; @@ -169,8 +169,29 @@ where timestamp_to_op: UnsafeCell>, headers: UnsafeCell>, inner: UnsafeCell>, + /// Ring of recently evicted committed entries, keyed by op, retained so + /// this replica can serve journal repair for rejoin windows after the + /// entries left the resident journal at flush. Bounded by + /// [`EVICTED_RING_CAPACITY`]; requests older than the ring answer + /// `RangeEvicted` honestly. + evicted_ring: UnsafeCell>, + /// Running byte total of the buffers held by `evicted_ring`. + evicted_ring_bytes: Cell, + /// Single-replica groups have nobody to repair; retaining evicted + /// entries for them is pure memory waste. + repair_retention: Cell, } +/// How many evicted entries each partition retains for repair. Sized to +/// cover a few seconds of traffic around a node restart; anything older is +/// bulk-sync (phase 3) territory. +pub const EVICTED_RING_CAPACITY: usize = 4096; + +/// Byte ceiling for the evicted ring: the entry cap alone lets each +/// partition pin up to 4096 full-sized batches, which is unbounded in byte +/// terms across many partitions. Whichever cap trips first evicts. +pub const EVICTED_RING_BYTES_MAX: u64 = 16 * 1024 * 1024; + impl Default for PartitionJournal where S: Storage + Default, @@ -184,6 +205,9 @@ where inner: UnsafeCell::new(JournalInner { storage: S::default(), }), + evicted_ring: UnsafeCell::new(VecDeque::new()), + evicted_ring_bytes: Cell::new(0), + repair_retention: Cell::new(true), } } } @@ -249,6 +273,52 @@ impl PartitionJournalMemStorage { } impl PartitionJournal { + /// Entry bytes for `op`, from the resident journal or the evicted ring. + /// `None` when the op predates the ring (bulk-sync territory) or was + /// never journaled here. + /// Disable repair retention (single-replica groups: nobody to repair). + pub fn set_repair_retention(&self, enabled: bool) { + self.repair_retention.set(enabled); + if !enabled { + let ring = unsafe { &mut *self.evicted_ring.get() }; + ring.clear(); + self.evicted_ring_bytes.set(0); + } + } + + /// Resident (un-evicted) entry count; diagnostics only. + pub fn resident_count(&self) -> usize { + let op_to_storage_offset = unsafe { &*self.op_to_storage_offset.get() }; + op_to_storage_offset.len() + } + + pub fn repair_entry(&self, op: u64) -> Option { + { + let op_to_storage_offset = unsafe { &*self.op_to_storage_offset.get() }; + if let Some(&storage_offset) = op_to_storage_offset.get(&op) { + let inner = unsafe { &*self.inner.get() }; + return Some(inner.storage.read_at_sync(storage_offset)); + } + } + let ring = unsafe { &*self.evicted_ring.get() }; + ring.iter() + .find(|(ring_op, _)| *ring_op == op) + .map(|(_, entry)| entry.clone()) + } + + /// Oldest op this journal can still serve for repair (ring front, else + /// resident head), or `None` when it holds nothing at all. + pub fn repair_retained_from(&self) -> Option { + { + let ring = unsafe { &*self.evicted_ring.get() }; + if let Some((op, _)) = ring.front() { + return Some(*op); + } + } + let headers = unsafe { &*self.headers.get() }; + headers.first().map(|header| header.op) + } + /// Synchronous resident-range poll read. Never awaits (mem storage reads /// are pure memory copies), so a partition borrow held across it cannot span /// a scheduler yield. The poll path uses this; the disk tier, which does @@ -368,6 +438,12 @@ impl PartitionJournal { let inner = unsafe { &*self.inner.get() }; inner.storage.drain() }; + // Ops are positional against `headers` until the clear below; capture + // the evicted prefix's ops first so the ring stays op-addressable. + let evicted_ops: Vec = { + let headers = unsafe { &*self.headers.get() }; + headers.iter().take(count).map(|header| header.op).collect() + }; { let headers = unsafe { &mut *self.headers.get() }; @@ -380,7 +456,36 @@ impl PartitionJournal { timestamp_to_op.clear(); } - let retained: Vec = all_entries.into_iter().skip(count).collect(); + let mut all_entries = all_entries.into_iter(); + if self.repair_retention.get() { + let ring = unsafe { &mut *self.evicted_ring.get() }; + let mut ring_bytes = self.evicted_ring_bytes.get(); + for op in evicted_ops { + let Some(entry) = all_entries.next() else { + break; + }; + ring_bytes += entry.len() as u64; + ring.push_back((op, entry)); + while ring.len() > EVICTED_RING_CAPACITY + || (ring_bytes > EVICTED_RING_BYTES_MAX && ring.len() > 1) + { + if let Some((_, dropped)) = ring.pop_front() { + ring_bytes -= dropped.len() as u64; + } + } + } + self.evicted_ring_bytes.set(ring_bytes); + } else { + // Consume without retaining: the iterator itself must still + // advance past the evicted prefix so the retained tail below is + // aligned. + for _ in &evicted_ops { + if all_entries.next().is_none() { + break; + } + } + } + let retained: Vec = all_entries.collect(); let mut result = Vec::with_capacity(retained.len()); for entry in retained { let meta = self @@ -490,6 +595,9 @@ where timestamp_to_op: UnsafeCell::new(BTreeMap::new()), headers: UnsafeCell::new(Vec::new()), inner: UnsafeCell::new(JournalInner { storage }), + evicted_ring: UnsafeCell::new(VecDeque::new()), + evicted_ring_bytes: Cell::new(0), + repair_retention: Cell::new(true), } } @@ -505,18 +613,19 @@ where /// backup, so this is a single linear scan: drop ops below `from_op`, take /// while contiguous, stop at the first gap or past `commit_max`. pub fn committed_headers_from(&self, from_op: u64, commit_max: u64) -> Vec { - let headers = unsafe { &*self.headers.get() }; + // Walk by OP, not by append position: after a rejoin the journal + // interleaves live tail ops (which arrive while repair is still + // streaming) with repaired window ops, so append order is no longer + // op-ascending and a positional sequential scan would break at the + // first interleave boundary forever. let mut result = Vec::new(); - let mut expected = from_op; - for header in headers { - if header.op < from_op { - continue; - } - if header.op != expected || header.op > commit_max { + let mut op = from_op; + while op <= commit_max { + let Some(header) = self.header_by_op(op) else { break; - } - result.push(*header); - expected += 1; + }; + result.push(header); + op += 1; } result } diff --git a/core/partitions/src/lib.rs b/core/partitions/src/lib.rs index 64ae0d2571..0abd137d14 100644 --- a/core/partitions/src/lib.rs +++ b/core/partitions/src/lib.rs @@ -45,7 +45,7 @@ use server_common::Message; pub use server_common::send_messages2::{IggyMessage2, IggyMessage2Header, IggyMessages2}; pub use types::{ AppendResult, Fragment, PartitionOffsets, PartitionsConfig, PollFragments, PollQueryResult, - PollingArgs, PollingConsumer, SendMessagesResult, + PollingArgs, PollingConsumer, REPAIR_RETRY_TICKS, RepairSession, SendMessagesResult, }; /// Partition-level data plane operations. diff --git a/core/partitions/src/types.rs b/core/partitions/src/types.rs index 3f2caa8a92..11c7796648 100644 --- a/core/partitions/src/types.rs +++ b/core/partitions/src/types.rs @@ -203,6 +203,39 @@ impl Default for PartitionOffsets { } } +/// Ticks of a stalled repair stream tolerated before a re-request. +/// +/// Partition group ticks are ~10ms, so ~1s. Repair frames are +/// fire-and-forget over a lossy bus; a session with no retry wedges +/// forever on a single dropped frame. The remaining window is re-requested +/// from the serving peer. +pub const REPAIR_RETRY_TICKS: u32 = 100; + +/// One in-flight journal-repair stream for a partition group. +#[derive(Debug, Clone, Copy)] +pub struct RepairSession { + /// Fences stale repair frames from an earlier attempt. + pub nonce: u128, + /// Last op the stream is expected to serve (the frontier at request time). + pub to_op: u64, + /// Commit floor learned from `RangeEvicted { retained_from }`: + /// `retained_from - 1`. `None` until (unless) the serving peer reports a + /// truncated prefix. + pub floor: Option, + /// The peer serving this stream (re-request target on stall). + pub peer: u8, + /// Lowest `base_offset` among the repaired `SendMessages` batches: + /// where the served window begins in offset space. Compared against the + /// boot-recovered durable end when a commit floor arrives -- a window + /// starting above `recovered_durable_offset + 1` means ops below the + /// floor are neither locally durable nor repaired (state-transfer + /// territory), and the floor must be refused. + pub first_batch_offset: Option, + /// Ticks since the stream last made progress; at + /// [`REPAIR_RETRY_TICKS`] the remaining window is re-requested. + pub idle_ticks: u32, +} + /// Configuration for partition operations. /// /// Mirrors the relevant fields from the server's `PartitionConfig` and diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs index ab2fd4c1d9..a0ebbf1044 100644 --- a/core/server-ng/src/bootstrap.rs +++ b/core/server-ng/src/bootstrap.rs @@ -894,7 +894,10 @@ async fn shard_main( // Notifier install deferred until after tick handler wires below. let senders_for_notifier = senders.clone(); let metrics_for_notifier = shard_metrics.clone(); - let (shard, sessions) = build_shard_for_thread( + // Heap-pin like `shard_main` above: the builder future carries the whole + // shard construction state machine and outgrew clippy's `large_futures` + // cap; one allocation per shard startup. + let (shard, sessions) = Box::pin(build_shard_for_thread( shard_id, total_shards, config, @@ -905,7 +908,7 @@ async fn shard_main( inbox, shard_metrics, Arc::clone(&metadata_view), - ) + )) .await?; // Shard 0 owns the metadata consensus; publish its view so every shard's @@ -1582,18 +1585,20 @@ fn restore_metadata_consensus( consensus.set_view(header.view); } - // On a RESTART in a cluster (a non-empty WAL proves a prior life), rejoin - // as a backup: resuming primaryship from the WAL's (stale) view races the - // peers' election and can split leadership across planes -- the partition - // groups always rejoin as backups (no WAL), so a metadata-primary resume - // here leaves metadata led by this node while the partitions elect a - // peer, and clients (which follow the roster's single leader) then write - // to a partition backup. The WAL still restores state below; only the - // role is ceded. A FRESH boot (empty WAL) keeps the plain init: the - // cluster needs its view-0 primary to exist, and a single-replica - // cluster has no peer to defer to. + // On a RESTART in a cluster (a non-empty WAL proves a prior life), + // rejoin as a quorum-invisible backup and probe for the current view + // (`RequestStartView`): the view's primary answers with a `StartView`, + // the replica adopts it as a backup, and journal repair fills any WAL + // gap. A probing replica never resumes primaryship -- if this replica + // IS the current primary-by-index, its probe makes the backups elect + // past it. + // The probe re-broadcasts on its timeout, so it needs no live mesh at + // boot. A FRESH boot (empty WAL) keeps the plain init: the cluster + // needs its view-0 primary to exist, and a single-replica cluster has + // no peer to ask. if replica_count > 1 && restored_op > 0 { consensus.init_as_backup(); + consensus.begin_view_probe(); } else { consensus.init(); } @@ -1683,13 +1688,16 @@ async fn load_partition( ); // A recovered partition lost its consensus state with the process: the // partition journal is in-memory and segments carry no op numbers, so - // this replica cannot know the group's (op, commit). In a cluster it must - // not boot as the view-0 primary heartbeating `commit_min = 0`; join as a - // backup and let a peer that kept its journal win the election and repair - // this replica through `StartView`. Single-replica groups have no peer to - // defer to, so they keep the plain init. + // this replica cannot know the group's (op, commit). In a cluster it + // boots as a quorum-invisible backup and probes for the current view + // (`RequestStartView`): the view's primary answers with a `StartView`, + // journal repair fills the rejoin window, and the commit floor settles + // at the serving peer's retention point. The probe re-broadcasts on its + // timeout, so it needs no live mesh at boot. Single-replica groups + // have no peer to ask and keep the plain init. if replica_count > 1 { - consensus.init_recovering(); + consensus.init_as_backup(); + consensus.begin_view_probe(); } else { consensus.init(); } @@ -1733,6 +1741,14 @@ async fn load_partition( .max() .unwrap_or(0); partition.created_at = partition_metadata.created_at; + if partition + .log + .segments() + .iter() + .any(|segment| segment.size > IggyByteSize::default()) + { + partition.recovered_durable_offset = Some(current_offset); + } partition.offset.store(current_offset, Ordering::Release); partition .dirty_offset diff --git a/core/server-ng/src/partition_helpers.rs b/core/server-ng/src/partition_helpers.rs index b544e9b4b3..54b3dbdb62 100644 --- a/core/server-ng/src/partition_helpers.rs +++ b/core/server-ng/src/partition_helpers.rs @@ -451,6 +451,17 @@ pub async fn build_partition_fresh( let partition_id = namespace.partition_id(); validate_namespace_bounds(config, stream_id, topic_id, partition_id)?; + // Sampled BEFORE the hierarchy create: a pre-existing partition directory + // is the marker of a prior life (the .log inside may legitimately be + // empty -- committed-but-unflushed data dies with the journal), while a + // genuinely fresh create finds nothing. + let restarted = replica_count > 1 + && std::fs::metadata( + config + .system + .get_partition_path(stream_id, topic_id, partition_id), + ) + .is_ok(); create_partition_file_hierarchy(stream_id, topic_id, partition_id, config) .await .map_err(|source| { @@ -472,7 +483,21 @@ pub async fn build_partition_fresh( bus, LocalPipeline::new(), ); - consensus.init(); + // A partition directory that already holds segment bytes is a RESTART + // materialization, not a fresh create: this replica's group state died + // with the process, so claiming view-0 primaryship would heartbeat + // commit_min=0 at peers that hold the committed log (racing their + // election). Join as a quorum-invisible backup and probe for the + // current view instead; journal repair re-materializes the data from a + // peer, byte-identical by the deterministic-roll/replicated-ciphertext + // design. A truly fresh create keeps the plain init: every group needs + // its view-0 primary to exist. + if restarted { + consensus.init_as_backup(); + consensus.begin_view_probe(); + } else { + consensus.init(); + } let mut partition = IggyPartition::new(stats, consensus); partition.set_partition_dir(config.system.get_partition_path( diff --git a/core/server-ng/src/partition_reconciler.rs b/core/server-ng/src/partition_reconciler.rs index 5e8b71bc7c..69e15ac73f 100644 --- a/core/server-ng/src/partition_reconciler.rs +++ b/core/server-ng/src/partition_reconciler.rs @@ -243,6 +243,11 @@ struct PassCounters { /// Consumer-group offsets reclaimed for groups deleted while their topic /// survived (a bare `DeleteConsumerGroup`, not a topic/stream delete). cg_offsets_purged: usize, + /// Committed delete watermarks not yet fully enforced on local segments. + /// Counted so the pass does not arm the fast-skip: the pump can be + /// blocked by a consumer barrier or by a rejoin whose offsets land via + /// journal repair, and neither unblocking bumps `Streams::revision`. + trims_pending: usize, } impl PassCounters { @@ -254,6 +259,7 @@ impl PassCounters { + self.backoff_skipped + self.stale + self.cg_offsets_purged + + self.trims_pending } } @@ -294,7 +300,7 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool { reconcile_additions(ctx, target, &mut counters).await; reconcile_removals(ctx, &target_set, &mut counters).await; reconcile_consumer_group_offsets(ctx, &mut counters).await; - reconcile_segment_truncations(ctx); + reconcile_segment_truncations(ctx, &mut counters); reconcile_partition_purges(ctx); let local_set: AHashSet = @@ -771,8 +777,14 @@ fn shards_table_contains(ctx: &ReconcilerCtx, ns: IggyNamespace) -> bool { /// carrying a non-zero delete watermark, stage a pump-side trim to that offset. /// Idempotent — the pump no-ops once a partition is trimmed past the watermark, /// so a redundant pass triggered by an unrelated revision bump is harmless. -fn reconcile_segment_truncations(ctx: &ReconcilerCtx) { - let namespaces: Vec<_> = ctx.shard.plane.partitions().namespaces().copied().collect(); +/// A watermark whose enforcement is still incomplete (first local segment +/// starts below it) counts as pending work: the pump may be blocked by a +/// consumer barrier or by a rejoin whose offsets arrive via journal repair, +/// and neither unblocking bumps `Streams::revision`, so the pass must keep +/// the reconciler ticking until the layout converges. +fn reconcile_segment_truncations(ctx: &ReconcilerCtx, counters: &mut PassCounters) { + let partitions = ctx.shard.plane.partitions(); + let namespaces: Vec<_> = partitions.namespaces().copied().collect(); let streams = ctx.shard.plane.metadata().mux_stm.streams(); for namespace in namespaces { let watermark = streams.partition_delete_watermark( @@ -780,8 +792,16 @@ fn reconcile_segment_truncations(ctx: &ReconcilerCtx) { namespace.topic_id(), namespace.partition_id(), ); - if watermark > 0 { - ctx.shard.request_truncate_partition(namespace, watermark); + if watermark == 0 { + continue; + } + ctx.shard.request_truncate_partition(namespace, watermark); + let trimmed = partitions + .get_by_ns(&namespace) + .and_then(|partition| partition.log.segments().first()) + .is_none_or(|first| first.start_offset >= watermark); + if !trimmed { + counters.trims_pending += 1; } } } diff --git a/core/server_common/src/consensus_message.rs b/core/server_common/src/consensus_message.rs index 60db83366f..04516b22ec 100644 --- a/core/server_common/src/consensus_message.rs +++ b/core/server_common/src/consensus_message.rs @@ -18,7 +18,8 @@ use crate::iobuf::{Frozen, Owned}; use iggy_binary_protocol::{ Command2, CommitHeader, ConsensusError, ConsensusHeader, DoViewChangeHeader, GenericHeader, - Operation, PrepareHeader, PrepareOkHeader, RequestHeader, StartViewChangeHeader, + Operation, PrepareHeader, PrepareOkHeader, RepairPrepareHeader, RepairRangeReplyHeader, + RequestHeader, RequestPreparesHeader, RequestStartViewHeader, StartViewChangeHeader, StartViewHeader, }; use smallvec::SmallVec; @@ -225,7 +226,7 @@ where } let generic = self.as_generic(); - if generic.header().command != T::COMMAND { + if !T::accepts(generic.header().command) { return Err(ConsensusError::InvalidCommand { expected: T::COMMAND, found: generic.header().command, @@ -261,7 +262,7 @@ where } let generic = self.as_generic(); - if generic.header().command != T::COMMAND { + if !T::accepts(generic.header().command) { return Err(ConsensusError::InvalidCommand { expected: T::COMMAND, found: generic.header().command, @@ -500,6 +501,17 @@ pub enum MessageBag { DoViewChange(Message), StartView(Message), Commit(Message), + RequestStartView(Message), + RequestPrepares(Message), + /// A journaled prepare re-sent verbatim for repair (command byte + /// distinguishes it from a live `Prepare`: repair bypasses the view + /// fence and never acks). Stays typed as `RepairPrepare` through every + /// parse -- the router round-trips frames through generic bytes, so a + /// parse-time byte restore would surface as a live `Prepare` on the + /// second pass and die on the view fence. + RepairPrepare(Message), + /// `RepairDone` / `RangeEvicted` (one layout, two commands). + RepairRangeReply(Message), } impl MessageBag { @@ -513,6 +525,10 @@ impl MessageBag { Self::DoViewChange(message) => message.header().command, Self::StartView(message) => message.header().command, Self::Commit(message) => message.header().command, + Self::RequestStartView(message) => message.header().command, + Self::RequestPrepares(message) => message.header().command, + Self::RepairPrepare(message) => message.header().command(), + Self::RepairRangeReply(message) => message.header().command, } } @@ -526,6 +542,10 @@ impl MessageBag { Self::DoViewChange(message) => message.header().size(), Self::StartView(message) => message.header().size(), Self::Commit(message) => message.header().size(), + Self::RequestStartView(message) => message.header().size(), + Self::RequestPrepares(message) => message.header().size(), + Self::RepairPrepare(message) => message.header().size(), + Self::RepairRangeReply(message) => message.header().size(), } } @@ -539,6 +559,10 @@ impl MessageBag { Self::DoViewChange(message) => message.header().operation(), Self::StartView(message) => message.header().operation(), Self::Commit(message) => message.header().operation(), + Self::RequestStartView(message) => message.header().operation(), + Self::RequestPrepares(message) => message.header().operation(), + Self::RepairPrepare(message) => message.header().operation(), + Self::RepairRangeReply(message) => message.header().operation(), } } } @@ -568,6 +592,22 @@ where )), Command2::StartView => Ok(Self::StartView(value.try_into_typed::()?)), Command2::Commit => Ok(Self::Commit(value.try_into_typed::()?)), + Command2::RequestStartView => Ok(Self::RequestStartView( + value.try_into_typed::()?, + )), + Command2::RequestPrepares => Ok(Self::RequestPrepares( + value.try_into_typed::()?, + )), + // A repaired prepare is a stored PrepareHeader frame whose command + // byte was rewritten; typed validation would reject the byte, so + // parse through the generic backing and trust the prepare-shaped + // layout the way the journal that produced it did. + Command2::RepairPrepare => Ok(Self::RepairPrepare( + value.try_into_typed::()?, + )), + Command2::RepairDone | Command2::RangeEvicted => Ok(Self::RepairRangeReply( + value.try_into_typed::()?, + )), // Reply / Eviction are server-to-client frames; they do not // appear on the inbound dispatch path. Command2::Reply | Command2::Eviction => { @@ -613,6 +653,50 @@ mod tests { o } + // MessageBag round-trip for the probe + repair command family. Locks + // RangeEvicted delivery in particular: RepairDone and RangeEvicted share + // one header layout and BOTH must survive the typed parse -- a strict + // command match in `try_into_typed` silently dropped every RangeEvicted + // frame and with it the whole commit-floor path. + #[test] + fn probe_and_repair_commands_round_trip_into_bag() { + for command in [ + Command2::RequestStartView, + Command2::RequestPrepares, + Command2::RepairPrepare, + Command2::RepairDone, + Command2::RangeEvicted, + ] { + let mut owned = header_bytes(command, 256); + if command == Command2::RequestPrepares { + // validate() demands a non-empty 1-based range. + const FROM_OP_OFF: usize = + std::mem::offset_of!(iggy_binary_protocol::RequestPreparesHeader, from_op); + const TO_OP_OFF: usize = + std::mem::offset_of!(iggy_binary_protocol::RequestPreparesHeader, to_op); + let buf = owned.as_mut_slice(); + buf[FROM_OP_OFF..FROM_OP_OFF + 8].copy_from_slice(&1u64.to_le_bytes()); + buf[TO_OP_OFF..TO_OP_OFF + 8].copy_from_slice(&1u64.to_le_bytes()); + } + let generic = Message::::try_from(owned) + .unwrap_or_else(|e| panic!("{command:?} failed generic framing: {e}")); + let bag = MessageBag::try_from(generic) + .unwrap_or_else(|e| panic!("{command:?} failed bag parse: {e}")); + let routed = matches!( + (&bag, command), + (MessageBag::RequestStartView(_), Command2::RequestStartView) + | (MessageBag::RequestPrepares(_), Command2::RequestPrepares) + | (MessageBag::RepairPrepare(_), Command2::RepairPrepare) + | ( + MessageBag::RepairRangeReply(_), + Command2::RepairDone | Command2::RangeEvicted + ) + ); + assert!(routed, "{command:?} parsed into the wrong bag variant"); + assert_eq!(bag.command(), command, "original command byte must survive"); + } + } + // Construction via Message::new (zeroed) #[test] diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 42857e723a..aff5089c72 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -32,7 +32,8 @@ use consensus::{ }; use iggy_binary_protocol::{ Command2, CommitHeader, DoViewChangeHeader, GenericHeader, Operation, PrepareHeader, - PrepareOkHeader, RequestHeader, StartViewChangeHeader, StartViewHeader, + PrepareOkHeader, RepairPrepareHeader, RepairRangeReplyHeader, RequestHeader, + RequestPreparesHeader, RequestStartViewHeader, StartViewChangeHeader, StartViewHeader, }; #[cfg(any(test, feature = "simulator"))] use iggy_common::PartitionStats; @@ -664,6 +665,26 @@ impl ShardFrame { } } +/// Prepares served per `RequestPrepares` round. The per-peer bus queues +/// are bounded (`peer_queue_capacity`, 256 by default) and overrun frames +/// drop silently, so an unbounded burst loses its own tail; the receiver +/// pulls the window chunk by chunk instead (each walked `RepairDone` +/// immediately requests the next chunk while progress holds). +const REPAIR_CHUNK_MAX: u64 = 128; + +/// One in-flight metadata journal-repair stream (shard 0 only). +#[derive(Debug, Clone, Copy)] +struct MetadataRepairSession { + nonce: u128, + to_op: u64, + /// Re-request target on stall. + peer: u8, + /// Ticks since the stream last made progress; at + /// [`partitions::REPAIR_RETRY_TICKS`] the remaining window is + /// re-requested from `peer`. + idle_ticks: u32, +} + pub struct IggyShard where B: MessageBus, @@ -686,6 +707,12 @@ where /// this shard. Invoked for each inbound `Request` frame. on_client_request: RequestHandler, + /// In-flight metadata journal repair: set when the recovery + /// handshake finds this replica's WAL behind the group frontier, cleared + /// at `RepairDone`. Metadata never needs a commit floor -- its WAL keeps + /// the full prefix -- so only the stream identity is tracked. + metadata_repair: RefCell>, + /// Handler for inbound [`MetadataSubmit`] frames. Only shard 0 receives /// these (it owns the metadata consensus group); peers send them here /// via [`Self::forward_metadata_submit`]. Defaults to a no-op for the @@ -832,6 +859,7 @@ where metadata_tick_handler: RefCell::new(None), reconcile_queue: RefCell::new(VecDeque::new()), pending_partition_frames: RefCell::new(HashMap::new()), + metadata_repair: RefCell::new(None), }) } @@ -1041,6 +1069,7 @@ where metadata_tick_handler: RefCell::new(None), reconcile_queue: RefCell::new(VecDeque::new()), pending_partition_frames: RefCell::new(HashMap::new()), + metadata_repair: RefCell::new(None), } } @@ -1337,6 +1366,10 @@ where Ok(MessageBag::DoViewChange(msg)) => self.on_do_view_change(msg).await, Ok(MessageBag::StartView(msg)) => self.on_start_view(msg).await, Ok(MessageBag::Commit(ref msg)) => self.on_commit(msg).await, + Ok(MessageBag::RequestStartView(ref msg)) => self.on_request_start_view(msg).await, + Ok(MessageBag::RequestPrepares(ref msg)) => self.on_request_prepares(msg).await, + Ok(MessageBag::RepairPrepare(msg)) => self.on_repair_prepare(msg).await, + Ok(MessageBag::RepairRangeReply(ref msg)) => self.on_repair_range_reply(msg).await, Err(e) => { tracing::warn!(shard = self.id, error = %e, "dropping message with invalid command"); } @@ -1752,6 +1785,40 @@ where { planes.0.commit_journal().await; } + // Adoption can leave this replica knowing a frontier its WAL + // cannot reach (StartView carries numbers, not entries): the + // walk above gap-stops. Fill the hole through journal repair + // from the announcing primary. + if consensus.is_normal() + && consensus.commit_min() < consensus.commit_max() + && self.metadata_repair.borrow().is_none() + { + let nonce = iggy_common::random_id::get_uuid(); + let to_op = consensus.commit_max(); + let from_op = consensus.commit_min() + 1; + *self.metadata_repair.borrow_mut() = Some(MetadataRepairSession { + nonce, + to_op, + peer: header.replica, + idle_ticks: 0, + }); + tracing::info!( + shard = self.id, + from_op, + to_op, + "metadata behind after StartView adoption; requesting repair" + ); + self.send_request_prepares( + consensus.cluster(), + consensus.replica(), + header.replica, + nonce, + from_op, + to_op, + header.namespace, + ) + .await; + } return; } @@ -1775,6 +1842,38 @@ where { partition.commit_journal(config).await; } + // Same gap-fill as the metadata arm: a journal-less rejoiner that + // adopted the new view still lacks the window's entries; repair from + // the announcing primary, floor settled by its RangeEvicted. + let consensus = partition.consensus(); + if consensus.is_normal() + && consensus.commit_min() < consensus.commit_max() + && partition.repair.is_none() + { + let nonce = iggy_common::random_id::get_uuid(); + let to_op = consensus.commit_max(); + let from_op = consensus.commit_min() + 1; + let cluster = consensus.cluster(); + let self_id = consensus.replica(); + partition.repair = Some(partitions::RepairSession { + nonce, + to_op, + floor: None, + peer: header.replica, + first_batch_offset: None, + idle_ticks: 0, + }); + self.send_request_prepares( + cluster, + self_id, + header.replica, + nonce, + from_op, + to_op, + header.namespace, + ) + .await; + } } #[allow(clippy::future_not_send)] @@ -1830,6 +1929,551 @@ where } } + /// `RequestStartView` probe from a restarted peer: the probed group's + /// current primary answers with a `StartView`; a probe from the replica + /// that IS the current primary-by-index makes backups elect immediately + /// (the consensus handler decides; everyone else stays silent). + #[allow(clippy::future_not_send)] + async fn on_request_start_view(&self, msg: &Message) + where + B: MessageBus, + MJ: JournalHandle, + ::Target: Journal< + ::Storage, + Entry = Message, + Header = PrepareHeader, + >, + { + let header = *msg.header(); + let planes = self.plane.inner(); + if let Some(ref consensus) = planes.0.consensus + && consensus.namespace() == header.namespace + { + let actions = consensus.handle_request_start_view(PlaneKind::Metadata, &header); + dispatch_vsr_actions(consensus, planes.0.journal.as_ref(), &actions).await; + return; + } + let Some(partition) = planes + .1 + .0 + .get_mut_by_ns(&IggyNamespace::from_raw(header.namespace)) + else { + return; + }; + let consensus = partition.consensus(); + let actions = consensus.handle_request_start_view(PlaneKind::Partitions, &header); + dispatch_vsr_actions::(consensus, None, &actions).await; + } + + /// Serve a repair range from this replica's journal: stream + /// `RepairPrepare` frames (stored prepares verbatim, command byte + /// rewritten) in op order, prefixed by `RangeEvicted` when the front of + /// the range is no longer retained, terminated by `RepairDone`. + #[allow(clippy::future_not_send, clippy::too_many_lines)] + async fn on_request_prepares(&self, msg: &Message) + where + B: MessageBus, + MJ: JournalHandle, + ::Target: Journal< + ::Storage, + Entry = Message, + Header = PrepareHeader, + >, + { + let header = *msg.header(); + let target = header.replica; + let planes = self.plane.inner(); + if let Some(ref consensus) = planes.0.consensus + && consensus.namespace() == header.namespace + { + if !consensus.is_normal() { + return; + } + let Some(journal) = planes.0.journal.as_ref() else { + return; + }; + let journal = journal.handle(); + let cluster = consensus.cluster(); + let self_id = consensus.replica(); + let to_op = header.to_op.min(consensus.commit_max()); + // Skip the compacted prefix (below the snapshot floor) in one + // RangeEvicted notice, then serve contiguously until the range + // ends or the WAL runs out. + let mut from_op = header.from_op; + #[allow(clippy::cast_possible_truncation)] + while from_op <= to_op && journal.header(from_op as usize).is_none() { + from_op += 1; + } + if from_op > to_op { + // Nothing in the requested range is retained. Answer the + // eviction honestly: a bare `RepairDone(to_op)` here would + // claim full coverage while serving zero prepares, and the + // requester would clear its session and gap-stop silently. + self.send_repair_range_reply( + cluster, + self_id, + target, + Command2::RangeEvicted, + header.nonce, + from_op, + header.namespace, + ) + .await; + self.send_repair_range_reply( + cluster, + self_id, + target, + Command2::RepairDone, + header.nonce, + header.from_op.saturating_sub(1), + header.namespace, + ) + .await; + return; + } + if from_op > header.from_op { + self.send_repair_range_reply( + cluster, + self_id, + target, + Command2::RangeEvicted, + header.nonce, + from_op, + header.namespace, + ) + .await; + } + let chunk_end = to_op.min(from_op.saturating_add(REPAIR_CHUNK_MAX - 1)); + let mut served_through = from_op.saturating_sub(1); + for op in from_op..=chunk_end { + #[allow(clippy::cast_possible_truncation)] + let Some(entry_header) = journal.header(op as usize).map(|h| *h) else { + break; + }; + let Some(entry) = journal.entry(&entry_header).await else { + break; + }; + if !self + .send_repair_prepare(target, entry.into_generic().into_frozen()) + .await + { + break; + } + served_through = op; + } + self.send_repair_range_reply( + cluster, + self_id, + target, + Command2::RepairDone, + header.nonce, + served_through, + header.namespace, + ) + .await; + return; + } + let Some(partition) = planes + .1 + .0 + .get_mut_by_ns(&IggyNamespace::from_raw(header.namespace)) + else { + return; + }; + if !partition.consensus().is_normal() { + return; + } + let cluster = partition.consensus().cluster(); + let self_id = partition.consensus().replica(); + let to_op = header.to_op.min(partition.consensus().commit_max()); + let retained_from = partition.log.journal().inner.repair_retained_from(); + let mut from_op = header.from_op; + if let Some(retained_from) = retained_from + && retained_from > from_op + { + self.send_repair_range_reply( + cluster, + self_id, + target, + Command2::RangeEvicted, + header.nonce, + retained_from, + header.namespace, + ) + .await; + from_op = retained_from; + } + let chunk_end = to_op.min(from_op.saturating_add(REPAIR_CHUNK_MAX - 1)); + let mut served_through = from_op.saturating_sub(1); + for op in from_op..=chunk_end { + let Some(entry) = partition.log.journal().inner.repair_entry(op) else { + break; + }; + if !self.send_repair_prepare(target, entry).await { + break; + } + served_through = op; + } + self.send_repair_range_reply( + cluster, + self_id, + target, + Command2::RepairDone, + header.nonce, + served_through, + header.namespace, + ) + .await; + tracing::info!( + shard = self.id, + namespace_raw = header.namespace, + target, + from_op = header.from_op, + to_op, + served_through, + "served partition repair range" + ); + } + + /// Ingest one repaired prepare. Metadata journals it into the WAL (the + /// commit walk at `RepairDone` applies it); partitions journal + stage it + /// through the same apply path as live replication, minus fence and ack. + #[allow(clippy::future_not_send)] + async fn on_repair_prepare(&self, msg: Message) + where + B: MessageBus, + MJ: JournalHandle, + ::Target: Journal< + ::Storage, + Entry = Message, + Header = PrepareHeader, + >, + { + tracing::debug!( + shard = self.id, + op = msg.header().0.op, + namespace_raw = msg.header().0.namespace, + "repair prepare received" + ); + // Convert to a live-prepare frame exactly once, here at the apply + // site (the frame must stay `RepairPrepare` up to this point: the + // router round-trips bags through generic bytes, and a live-Prepare + // command byte would land the re-parse on the view fence). The + // inner layout IS a stored prepare; downstream journal/apply paths + // run full prepare validation on it. + let msg = msg.transmute_header(|old: RepairPrepareHeader, new: &mut PrepareHeader| { + *new = old.0; + new.command = Command2::Prepare; + }); + let header = *msg.header(); + let planes = self.plane.inner(); + if let Some(ref consensus) = planes.0.consensus + && consensus.namespace() == header.namespace + { + let session = *self.metadata_repair.borrow(); + let Some(session) = session else { + return; + }; + if header.op > session.to_op || header.op <= consensus.commit_min() { + return; + } + let Some(journal) = planes.0.journal.as_ref() else { + return; + }; + let journal = journal.handle(); + #[allow(clippy::cast_possible_truncation)] + if journal.header(header.op as usize).is_some() { + return; + } + if let Err(error) = journal.append(msg).await { + tracing::warn!( + shard = self.id, + op = header.op, + %error, + "failed to journal repaired metadata prepare" + ); + return; + } + // Contiguous-frontier advance, mirroring + // `apply_repaired_prepare`: DVC advertises the sequencer, so a + // hole below a repaired op must stall the advance rather than + // mint an election candidate with an unwalkable log. + let mut frontier = consensus.sequencer().current_sequence(); + #[allow(clippy::cast_possible_truncation)] + while journal.header((frontier + 1) as usize).is_some() { + frontier += 1; + } + if frontier > consensus.sequencer().current_sequence() { + consensus.sequencer().set_sequence(frontier); + } + consensus.set_last_prepare_checksum(header.checksum); + return; + } + let Some(partition) = planes + .1 + .0 + .get_mut_by_ns(&IggyNamespace::from_raw(header.namespace)) + else { + return; + }; + partition.apply_repaired_prepare(msg).await; + } + + /// Repair stream terminator: `RangeEvicted` settles the partition commit + /// floor candidate; `RepairDone` runs the commit walk over the repaired + /// window and closes the session. + #[allow(clippy::future_not_send, clippy::too_many_lines)] + async fn on_repair_range_reply(&self, msg: &Message) + where + B: MessageBus, + MJ: JournalHandle, + ::Target: Journal< + ::Storage, + Entry = Message, + Header = PrepareHeader, + >, + M: StreamsFrontend + + StateMachine< + Input = Message, + Output = metadata::stm::result::ApplyReply, + Error = iggy_common::IggyError, + >, + { + let header = *msg.header(); + let planes = self.plane.inner(); + if let Some(ref consensus) = planes.0.consensus + && consensus.namespace() == header.namespace + { + let session = *self.metadata_repair.borrow(); + let Some(session) = session else { + return; + }; + if header.nonce != session.nonce { + return; + } + match header.command { + Command2::RepairDone => { + let before = consensus.commit_min(); + planes.0.commit_journal().await; + // Completion is decided by the LOCAL walk, not the + // peer's served-through claim: repair frames ride a + // lossy best-effort bus, so a fully-served stream can + // still arrive with holes. Anything short keeps the + // session armed; while the walk is making progress the + // next chunk is pulled immediately (the window is served + // in `REPAIR_CHUNK_MAX` slices), and a stalled one is + // left to the retry timer. + let commit_min = consensus.commit_min(); + let done = commit_min >= session.to_op; + tracing::info!( + shard = self.id, + through_op = header.op, + commit_min, + done, + "metadata journal repair walked" + ); + if done { + *self.metadata_repair.borrow_mut() = None; + } else if commit_min > before { + self.send_request_prepares( + consensus.cluster(), + consensus.replica(), + session.peer, + session.nonce, + commit_min + 1, + session.to_op, + header.namespace, + ) + .await; + } + } + Command2::RangeEvicted => { + // The metadata WAL retains everything above the snapshot + // floor; an evicted range means the peer compacted past + // this replica's gap -- bulk snapshot transfer (phase 3) + // territory. Surface loudly; the divergence backstop + // still guards the walk. + tracing::warn!( + shard = self.id, + retained_from = header.op, + "metadata repair range evicted on the serving peer; \ + snapshot transfer required" + ); + } + _ => {} + } + return; + } + let config = planes.1.0.config().clone(); + let Some(partition) = planes + .1 + .0 + .get_mut_by_ns(&IggyNamespace::from_raw(header.namespace)) + else { + return; + }; + let Some(session) = partition.repair else { + return; + }; + if header.nonce != session.nonce { + return; + } + match header.command { + Command2::RangeEvicted => { + if let Some(repair) = partition.repair.as_mut() { + repair.floor = Some(header.op.saturating_sub(1)); + } + } + Command2::RepairDone => { + // `complete_repair` walks the window and clears the session + // only when the LOCAL commit frontier reached the requested + // op (the peer's served-through claim proves nothing about + // delivery on a lossy bus). While the walk makes progress + // the next chunk is pulled immediately; a stalled window is + // left to the retry timer. + let before = partition.consensus().commit_min(); + partition.complete_repair(&config).await; + if partition.repair.is_none() { + tracing::info!( + shard = self.id, + namespace_raw = header.namespace, + through_op = header.op, + "partition journal repair complete" + ); + } else { + let commit_min = partition.consensus().commit_min(); + let next = partition.repair.as_ref().and_then(|live| { + (commit_min > before).then_some((live.peer, live.nonce, live.to_op)) + }); + let cluster = partition.consensus().cluster(); + let self_id = partition.consensus().replica(); + if let Some((peer, nonce, to_op)) = next { + self.send_request_prepares( + cluster, + self_id, + peer, + nonce, + commit_min + 1, + to_op, + header.namespace, + ) + .await; + } + } + } + _ => {} + } + } + + /// Ask `target` to stream its journaled prepares in `[from_op, to_op]` + /// for `namespace`; answered by `RepairPrepare` frames terminated with + /// `RepairDone` (prefixed by `RangeEvicted` when the front of the range + /// is no longer retained). + #[allow(clippy::too_many_arguments)] + #[allow(clippy::future_not_send, clippy::cast_possible_truncation)] + async fn send_request_prepares( + &self, + cluster: u128, + self_id: u8, + target: u8, + nonce: u128, + from_op: u64, + to_op: u64, + namespace: u64, + ) where + B: MessageBus, + { + let msg = Message::::new(size_of::()) + .transmute_header(|_, h: &mut RequestPreparesHeader| { + h.command = Command2::RequestPrepares; + h.cluster = cluster; + h.replica = self_id; + h.nonce = nonce; + h.from_op = from_op; + h.to_op = to_op; + h.namespace = namespace; + h.size = size_of::() as u32; + }); + if self + .bus + .send_to_replica(target, msg.into_generic().into_frozen()) + .await + .is_err() + { + // The stall retry re-requests; without this line a dead peer + // channel makes repair look like a silent server-side refusal. + tracing::warn!( + shard = self.id, + target, + from_op, + to_op, + namespace_raw = namespace, + "request-prepares send failed; stall retry will re-request" + ); + } + } + + /// Send a stored prepare (raw journal bytes) as a `RepairPrepare` frame: + /// the command byte is rewritten on an owned copy, because a verbatim + /// `Prepare` would hit the live view fence on the receiver while + /// `RepairPrepare` routes to the fence-free repair ingest. + /// Returns whether the frame was handed to the bus: `send_to_replica` + /// is a non-blocking try-send, so under a queue-full burst op N can be + /// dropped while N+1 lands. Callers must not advance their + /// served-through watermark past a failed send, or the terminating + /// `RepairDone` reports ops that were never delivered. + #[allow(clippy::future_not_send)] + async fn send_repair_prepare(&self, target: u8, entry: Frozen) -> bool + where + B: MessageBus, + { + const COMMAND_OFFSET: usize = std::mem::offset_of!(GenericHeader, command); + let mut owned = + server_common::iobuf::Owned::::copy_from_slice(entry.as_slice()); + owned.as_mut_slice()[COMMAND_OFFSET] = Command2::RepairPrepare as u8; + let Ok(message) = Message::::try_from(owned) else { + tracing::warn!( + shard = self.id, + "repair prepare bytes failed message framing" + ); + return false; + }; + self.bus + .send_to_replica(target, message.into_frozen()) + .await + .is_ok() + } + + #[allow(clippy::too_many_arguments)] + #[allow(clippy::future_not_send, clippy::cast_possible_truncation)] + async fn send_repair_range_reply( + &self, + cluster: u128, + self_id: u8, + target: u8, + command: Command2, + nonce: u128, + op: u64, + namespace: u64, + ) where + B: MessageBus, + { + let msg = Message::::new(size_of::()) + .transmute_header(|_, h: &mut RepairRangeReplyHeader| { + h.command = command; + h.cluster = cluster; + h.replica = self_id; + h.nonce = nonce; + h.op = op; + h.namespace = namespace; + h.size = size_of::() as u32; + }); + let _ = self + .bus + .send_to_replica(target, msg.into_generic().into_frozen()) + .await; + } + /// Tick partition consensuses. Loop partitions. No partitions-plane journal. #[allow(clippy::future_not_send)] pub async fn tick_partitions(&self) @@ -1862,6 +2506,61 @@ where let actions = consensus.tick(PlaneKind::Partitions); dispatch_vsr_actions::(consensus, None, &actions).await; dispatch_partition_journal_actions(consensus, partition, &actions).await; + + // Stall retry: repair frames are fire-and-forget, so a lost + // frame (or a peer that went silent mid-stream) would leave the + // session armed forever with commit_min pinned below commit_max. + // Re-request the remaining window from the serving peer; the + // ingest path skips duplicates, so overlap is harmless. + let stalled = { + let Some(partition) = partitions.get_mut_by_ns(&namespace) else { + continue; + }; + let consensus_normal = partition.consensus().is_normal(); + let commit_min = partition.consensus().commit_min(); + let cluster = partition.consensus().cluster(); + let self_id = partition.consensus().replica(); + partition.repair.as_mut().and_then(|session| { + if !consensus_normal { + return None; + } + session.idle_ticks += 1; + if session.idle_ticks < partitions::REPAIR_RETRY_TICKS { + return None; + } + session.idle_ticks = 0; + Some(( + session.peer, + session.nonce, + commit_min + 1, + session.to_op, + cluster, + self_id, + )) + }) + }; + if let Some((peer, nonce, from_op, to_op, cluster, self_id)) = stalled + && from_op <= to_op + { + tracing::info!( + shard = self.id, + namespace_raw = namespace.inner(), + from_op, + to_op, + peer, + "partition repair stalled; re-requesting remaining window" + ); + self.send_request_prepares( + cluster, + self_id, + peer, + nonce, + from_op, + to_op, + namespace.inner(), + ) + .await; + } } } @@ -1924,6 +2623,45 @@ where // forever (commit_min stuck below commit_max). See // `IggyMetadata::repair_primary_self_acks`. metadata.repair_primary_self_acks().await; + + // Stall retry, mirroring `tick_partitions`: a lost repair frame must + // not wedge the session forever. + let stalled = { + let mut session = self.metadata_repair.borrow_mut(); + session.as_mut().and_then(|session| { + if !consensus.is_normal() { + return None; + } + session.idle_ticks += 1; + if session.idle_ticks < partitions::REPAIR_RETRY_TICKS { + return None; + } + session.idle_ticks = 0; + Some((session.peer, session.nonce, session.to_op)) + }) + }; + if let Some((peer, nonce, to_op)) = stalled { + let from_op = consensus.commit_min() + 1; + if from_op <= to_op { + tracing::info!( + shard = self.id, + from_op, + to_op, + peer, + "metadata repair stalled; re-requesting remaining window" + ); + self.send_request_prepares( + consensus.cluster(), + consensus.replica(), + peer, + nonce, + from_op, + to_op, + consensus.namespace(), + ) + .await; + } + } } } @@ -1957,6 +2695,23 @@ where dispatch_vsr_actions::(consensus, None, &[action]).await; } +/// Re-stamp a stored prepare with the current view before retransmission. +/// After a view change the primary re-sends its uncommitted suffix as its +/// own prepares (VSR), but the journal keeps the original view stamp and +/// `replicate_preflight` fences `header.view < view` as deposed-primary +/// traffic -- a verbatim replay of the stored bytes would be ignored +/// forever, wedging the commit walk on every peer. The stored buffer is +/// shared with the journal, so the patch runs on an owned copy. +fn restamp_prepare_view(stored: &[u8], view: u32) -> Option> { + const VIEW_OFFSET: usize = std::mem::offset_of!(PrepareHeader, view); + let mut owned = server_common::iobuf::Owned::::copy_from_slice(stored); + owned.as_mut_slice()[VIEW_OFFSET..VIEW_OFFSET + std::mem::size_of::()] + .copy_from_slice(&view.to_ne_bytes()); + Message::::try_from(owned) + .ok() + .map(Message::into_frozen) +} + /// Dispatch a list of `VsrAction`s by constructing the appropriate /// protocol messages and sending them via the consensus message bus. #[allow( @@ -2037,6 +2792,19 @@ async fn dispatch_vsr_actions( }); send(*target, msg.into_generic().into_frozen()).await; } + VsrAction::SendRequestStartView { view, namespace } => { + let msg = + Message::::new(size_of::()) + .transmute_header(|_, h: &mut RequestStartViewHeader| { + h.command = Command2::RequestStartView; + h.cluster = cluster; + h.replica = self_id; + h.view = *view; + h.namespace = *namespace; + h.size = size_of::() as u32; + }); + broadcast(msg.into_generic().into_frozen()).await; + } VsrAction::SendStartView { view, op, @@ -2094,12 +2862,22 @@ async fn dispatch_vsr_actions( let Some(journal) = journal else { continue; }; + let current_view = consensus.view(); for (header, replicas) in targets { let Some(prepare) = journal.handle().entry(header).await else { continue; }; // Freeze the retransmit payload once; clone per target. - let frozen = prepare.into_generic().into_frozen(); + let frozen = if prepare.header().view == current_view { + prepare.into_generic().into_frozen() + } else { + let Some(restamped) = + restamp_prepare_view(prepare.as_slice(), current_view) + else { + continue; + }; + restamped + }; for replica in replicas { send(*replica, frozen.clone()).await; } @@ -2124,10 +2902,11 @@ async fn dispatch_vsr_actions( }) .collect(); if let Some(missing_op) = gap_at { - // Journal repair is not yet implemented.Truncate the sequencer - // to the last op we could rebuild so the next client - // prepare chains correctly. Ops above the - // gap are lost until journal repair is added. + // A primary's own uncommitted suffix has no repair + // source: peers ack'd nothing above the gap or the DVC + // merge would have carried it, so the range is decided + // lost. Truncate the sequencer to the last op we could + // rebuild so the next client prepare chains correctly. let rebuilt_up_to = missing_op.saturating_sub(1); tracing::warn!( replica = self_id, @@ -2244,6 +3023,7 @@ async fn dispatch_partition_journal_actions( // needs to become durable before cluster workloads go to // production. Server boot emits a loud warning to the // operator (see `main.rs`). + let current_view = consensus.view(); for (header, replicas) in targets { let Some(prepare) = journal.entry(header).await else { continue; @@ -2255,6 +3035,16 @@ async fn dispatch_partition_journal_actions( // above and avoids both the per-target 4 KiB memcpy // and the prior `.expect` that would panic the shard // on a corrupted journal entry. + let prepare = if header.view == current_view { + prepare + } else { + let Some(restamped) = + restamp_prepare_view(prepare.as_slice(), current_view) + else { + continue; + }; + restamped + }; for replica in replicas { send(*replica, prepare.clone()).await; } diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs index 90ea1a280d..4d457126f9 100644 --- a/core/shard/src/router.rs +++ b/core/shard/src/router.rs @@ -71,6 +71,22 @@ fn extract_routing(bag: MessageBag) -> (Operation, u64, Message) let h = *m.header(); (h.operation(), h.namespace, m.into_generic()) } + MessageBag::RequestStartView(m) => { + let h = *m.header(); + (h.operation(), h.namespace, m.into_generic()) + } + MessageBag::RequestPrepares(m) => { + let h = *m.header(); + (h.operation(), h.namespace, m.into_generic()) + } + MessageBag::RepairPrepare(m) => { + let h = *m.header(); + (h.0.operation, h.0.namespace, m.into_generic()) + } + MessageBag::RepairRangeReply(m) => { + let h = *m.header(); + (h.operation(), h.namespace, m.into_generic()) + } } }