This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch topic_durability in repository https://gitbox.apache.org/repos/asf/iggy.git
commit df0d49ff7a481c61656e84ebc36c4904a3f4f18b Author: Hubert Gruszecki <[email protected]> AuthorDate: Thu Sep 10 21:21:04 2026 +0200 fixes --- Cargo.lock | 1 + core/bench/src/args/examples.rs | 5 +- .../kinds/balanced/producer_and_consumer_group.rs | 4 +- .../kinds/end_to_end/producing_consumer_group.rs | 2 +- core/bench/src/args/kinds/pinned/consumer.rs | 2 +- core/bench/src/utils/mod.rs | 50 ++++- core/consensus/src/impls.rs | 175 +++++++++++----- core/consensus/src/plane_helpers.rs | 227 +++++++++++++++++++-- core/integration/src/harness/disk.rs | 13 +- .../tests/cluster/crash_recovery_corruption.rs | 36 ++-- core/journal/Cargo.toml | 3 + core/journal/src/durable_storage.rs | 6 +- core/journal/src/partition_journal.rs | 155 ++++++++++++-- core/journal/src/partition_journal/segments.rs | 47 +++-- core/metadata/src/stm/stream.rs | 9 +- core/partitions/src/iggy_partition.rs | 185 +++++++++++++++-- core/partitions/src/offset_storage.rs | 4 + core/partitions/src/persistence.rs | 24 +-- core/partitions/src/state_transfer.rs | 30 ++- core/server/config.toml | 6 +- core/server/src/dispatch/partition.rs | 108 ++++++++++ core/server/src/partition_helpers.rs | 93 ++++++++- core/server/src/responses.rs | 2 +- core/server/src/segment_recovery.rs | 4 +- core/server/src/server_error.rs | 36 +++- core/server_common/src/fs_utils.rs | 19 +- core/shard/src/lib.rs | 6 + core/shard/src/metrics.rs | 6 +- core/shard/src/router.rs | 11 +- core/simulator/src/replica.rs | 2 +- core/simulator/src/storage.rs | 13 +- core/simulator/src/storage/tests.rs | 108 ++++++++++ 32 files changed, 1181 insertions(+), 211 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 03dc277e3..007cf389c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8000,6 +8000,7 @@ dependencies = [ "futures", "iggy_binary_protocol", "iggy_common", + "nix", "server_common", "tempfile", "tracing", diff --git a/core/bench/src/args/examples.rs b/core/bench/src/args/examples.rs index f461e626a..7cb33b6e4 100644 --- a/core/bench/src/args/examples.rs +++ b/core/bench/src/args/examples.rs @@ -144,7 +144,7 @@ pub fn print_examples() { mod tests { use super::EXAMPLES; use crate::args::common::IggyBenchArgs; - use clap::{CommandFactory, Parser, error::ErrorKind}; + use clap::{CommandFactory, FromArgMatches, Parser, error::ErrorKind}; use iggy::prelude::Durability; use std::collections::BTreeSet; @@ -162,6 +162,9 @@ mod tests { let (kind, options) = matches.subcommand().unwrap(); kinds.insert(kind.to_owned()); transports.insert(options.subcommand_name().unwrap().to_owned()); + IggyBenchArgs::from_arg_matches(&matches) + .unwrap() + .validate(); } Err(error) => { assert_eq!(error.kind(), ErrorKind::DisplayHelp, "{command}: {error}"); diff --git a/core/bench/src/args/kinds/balanced/producer_and_consumer_group.rs b/core/bench/src/args/kinds/balanced/producer_and_consumer_group.rs index add19b461..cb2d7e178 100644 --- a/core/bench/src/args/kinds/balanced/producer_and_consumer_group.rs +++ b/core/bench/src/args/kinds/balanced/producer_and_consumer_group.rs @@ -107,7 +107,7 @@ impl BenchmarkKindProps for BalancedProducerAndConsumerGroupArgs { if cg_number < streams { cmd.error( ErrorKind::ArgumentConflict, - "Consumer groups number must be less than or equal to the number of streams.", + "Consumer groups number must be greater than or equal to the number of streams.", ) .exit(); } @@ -118,7 +118,7 @@ impl BenchmarkKindProps for BalancedProducerAndConsumerGroupArgs { if partitions < consumers { cmd.error( ErrorKind::ArgumentConflict, - "Consumer number must be greater than the number of partitions.", + "Consumer number must be less than or equal to the number of partitions.", ) .exit(); } diff --git a/core/bench/src/args/kinds/end_to_end/producing_consumer_group.rs b/core/bench/src/args/kinds/end_to_end/producing_consumer_group.rs index f67d550d6..bf935bae7 100644 --- a/core/bench/src/args/kinds/end_to_end/producing_consumer_group.rs +++ b/core/bench/src/args/kinds/end_to_end/producing_consumer_group.rs @@ -113,7 +113,7 @@ impl BenchmarkKindProps for EndToEndProducingConsumerGroupArgs { cmd.error( ErrorKind::ArgumentConflict, format!( - "For producing consumer group benchmark, consumer groups number ({cg_number}) must be less than the number of streams ({streams})" + "For producing consumer group benchmark, consumer groups number ({cg_number}) must be greater than or equal to the number of streams ({streams})" ), ) .exit(); diff --git a/core/bench/src/args/kinds/pinned/consumer.rs b/core/bench/src/args/kinds/pinned/consumer.rs index 12cdd4090..155d8cc54 100644 --- a/core/bench/src/args/kinds/pinned/consumer.rs +++ b/core/bench/src/args/kinds/pinned/consumer.rs @@ -74,7 +74,7 @@ impl BenchmarkKindProps for PinnedConsumerArgs { if streams > consumers { cmd.error( ErrorKind::ArgumentConflict, - format!("For pinned consumer, number of streams ({streams}) must be equal to the number of consumers ({consumers}).", + format!("For pinned consumer, number of streams ({streams}) must be less than or equal to the number of consumers ({consumers}).", )) .exit(); } diff --git a/core/bench/src/utils/mod.rs b/core/bench/src/utils/mod.rs index 3c9848c4a..671f6eca8 100644 --- a/core/bench/src/utils/mod.rs +++ b/core/bench/src/utils/mod.rs @@ -300,7 +300,8 @@ fn add_infrastructure_arguments(parts: &mut Vec<String>, args: &IggyBenchArgs) { let default_streams = match args.benchmark_kind.as_simple_kind() { BenchmarkKind::BalancedProducerAndConsumerGroup | BenchmarkKind::BalancedConsumerGroup - | BenchmarkKind::BalancedProducer => DEFAULT_BALANCED_NUMBER_OF_STREAMS.get(), + | BenchmarkKind::BalancedProducer + | BenchmarkKind::EndToEndProducingConsumerGroup => DEFAULT_BALANCED_NUMBER_OF_STREAMS.get(), _ => DEFAULT_PINNED_NUMBER_OF_STREAMS.get(), }; if streams != default_streams { @@ -311,10 +312,13 @@ fn add_infrastructure_arguments(parts: &mut Vec<String>, args: &IggyBenchArgs) { let default_partitions = match args.benchmark_kind.as_simple_kind() { BenchmarkKind::BalancedProducerAndConsumerGroup | BenchmarkKind::BalancedConsumerGroup - | BenchmarkKind::BalancedProducer => DEFAULT_BALANCED_NUMBER_OF_PARTITIONS.get(), + | BenchmarkKind::BalancedProducer + | BenchmarkKind::EndToEndProducingConsumerGroup => { + DEFAULT_BALANCED_NUMBER_OF_PARTITIONS.get() + } _ => DEFAULT_PINNED_NUMBER_OF_PARTITIONS.get(), }; - if partitions != default_partitions { + if partitions != 0 && partitions != default_partitions { parts.push(format!("--partitions {partitions}")); } @@ -362,6 +366,46 @@ mod tests { use crate::args::common::IggyBenchArgs; use clap::Parser; + #[test] + fn reproduced_commands_preserve_consumer_only_and_end_to_end_topologies() { + for arguments in [ + vec!["iggy-bench", "pinned-consumer", "tcp"], + vec!["iggy-bench", "balanced-consumer-group", "tcp"], + vec!["iggy-bench", "end-to-end-producing-consumer-group", "tcp"], + vec![ + "iggy-bench", + "end-to-end-producing-consumer-group", + "--streams", + "6", + "--consumer-groups", + "6", + "--partitions", + "1", + "tcp", + ], + ] { + let mut original = IggyBenchArgs::try_parse_from(arguments).unwrap(); + original.validate(); + let command = recreate_bench_command(&original); + let arguments = command + .split_ascii_whitespace() + .skip_while(|argument| *argument != "iggy-bench"); + let mut reproduced = IggyBenchArgs::try_parse_from(arguments).unwrap(); + reproduced.validate(); + assert_eq!(reproduced.streams(), original.streams(), "{command}"); + assert_eq!( + reproduced.number_of_partitions(), + original.number_of_partitions(), + "{command}" + ); + assert_eq!( + reproduced.number_of_consumer_groups(), + original.number_of_consumer_groups(), + "{command}" + ); + } + } + #[test] fn reproduced_websocket_commands_preserve_independent_topic_policies() { for messages in ["replicated", "persisted"] { diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index 0d1ebce94..4a2ae9785 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -2523,9 +2523,8 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { /// IS the whole quorum. `PrepareOk` loops through the loopback because it /// genuinely is a message to a peer that happens to be this replica. /// - /// The three callers that used to inline this sequence were the actual - /// duplication: an election timeout, an SVC for a higher view, and a DVC for - /// a higher view, differing only in `reason`. + /// A timed-out primary candidate can become a backup in the next view, + /// so every transition must restart its probe for a missing `StartView`. fn enter_view_change( &self, plane: PlaneKind, @@ -2640,43 +2639,11 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { return Vec::new(); } - // Escalate: try next view - let old_view = self.view.get(); - let next_view = old_view + 1; - - self.view.set(next_view); - self.reset_view_change_state(); - self.sent_own_start_view_change.set(true); - self.start_view_change_from_all_replicas - .borrow_mut() - .insert(self.replica as usize); - - self.timeouts - .borrow_mut() - .reset(TimeoutKind::ViewChangeStatus); - - emit_sim_event( - SimEventKind::ViewChangeStarted, - &ViewChangeLogEvent { - replica: ReplicaLogContext::from_consensus(self, plane), - old_view, - new_view: next_view, - reason: ViewChangeReason::ViewChangeStatusTimeout, - }, - ); - - let action = VsrAction::SendStartViewChange { - view: next_view, - group: self.group, - }; - emit_sim_event( - SimEventKind::ControlMessageScheduled, - &ControlActionLogEvent::from_vsr_action( - ReplicaLogContext::from_consensus(self, plane), - &action, - ), - ); - vec![action] + self.enter_view_change( + plane, + self.view.get() + 1, + ViewChangeReason::ViewChangeStatusTimeout, + ) } /// Collect uncommitted pipeline entries that should be retransmitted. @@ -3981,22 +3948,28 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { /// Send a message to `target`, routing self-addressed messages through the loopback queue. // VsrConsensus uses Cell/RefCell for single-threaded compio shards; futures are intentionally !Send. #[allow(clippy::future_not_send)] - pub(crate) async fn send_or_loopback(&self, target: u8, message: Message<GenericHeader>) + pub(crate) async fn send_or_loopback(&self, target: u8, message: Message<GenericHeader>) -> bool where B: MessageBus, { if target == self.replica { self.push_loopback(message); - } else if let Err(e) = self + return true; + } + match self .message_bus .send_to_replica(target, message.into_frozen()) .await { - tracing::warn!( - replica = self.replica, - target, - "send_or_loopback failed: {e}" - ); + Ok(()) => true, + Err(error) => { + tracing::warn!( + replica = self.replica, + target, + "send_or_loopback failed: {error}" + ); + false + } } } @@ -5024,6 +4997,114 @@ mod vsr_consensus_tests { ); } + #[test] + fn given_a_stalled_candidate_when_it_escalates_should_recover_a_lost_start_view() { + const REJOINING_REPLICA: u8 = 1; + const NEXT_PRIMARY: u8 = 2; + const REPLICA_COUNT: u8 = 3; + const LOCAL_HEAD: u64 = 20; + const PRIMARY_HEAD: u64 = LOCAL_HEAD + 1; + let plane = PlaneKind::Partitions; + let group = IggyNamespace::new(0, 0, 0).inner(); + let backup = VsrConsensus::new( + 1, + REJOINING_REPLICA, + REPLICA_COUNT, + group, + StageNoopBus, + LocalPipeline::new(), + ); + backup.sequencer().set_sequence(LOCAL_HEAD); + backup.restore_commit_state(LOCAL_HEAD - 1, LOCAL_HEAD - 1); + backup.begin_view_probe(); + for _ in 0..TimeoutManager::REQUEST_START_VIEW_MESSAGE_TICKS * u64::from(PROBE_ATTEMPTS_MAX) + { + backup.tick(plane); + } + assert_eq!(backup.view(), 1); + assert_eq!(backup.status(), Status::ViewChange); + assert!(backup.is_primary_for_view(backup.view())); + + for _ in 0..TimeoutManager::VIEW_CHANGE_STATUS_TICKS { + backup.tick(plane); + } + assert_eq!(backup.view(), 2); + assert_eq!(backup.status(), Status::ViewChange); + assert!(!backup.is_primary_for_view(backup.view())); + + // Model a settled primary whose initial StartView was withheld by persistence. + let mut primary = VsrConsensus::new( + 1, + NEXT_PRIMARY, + REPLICA_COUNT, + group, + StageNoopBus, + LocalPipeline::new(), + ); + primary.set_view(backup.view()); + primary.set_log_view(backup.view()); + primary.sequencer().set_sequence(PRIMARY_HEAD); + primary.restore_commit_state(LOCAL_HEAD, LOCAL_HEAD); + primary.init(); + + let probe = (0..TimeoutManager::REQUEST_START_VIEW_MESSAGE_TICKS) + .flat_map(|_| backup.tick(plane)) + .find_map(|action| match action { + VsrAction::SendRequestStartView { view, group } => Some( + Message::<RequestStartViewHeader>::new(size_of::<RequestStartViewHeader>()) + .transmute_header(|_, header: &mut RequestStartViewHeader| { + header.command = Command::RequestStartView; + header.cluster = 1; + header.replica = REJOINING_REPLICA; + header.view = view; + header.group = group; + header.size = + u32::try_from(size_of::<RequestStartViewHeader>()).unwrap(); + header.seal(); + }), + ), + _ => None, + }) + .expect("an escalated candidate must request the missing StartView as a backup"); + let replies = primary.handle_request_start_view(plane, probe.header()); + let [ + VsrAction::SendStartView { + view, + op, + commit, + incarnation, + target, + group, + suffix, + }, + ] = replies.as_slice() + else { + panic!("the settled primary must answer the backup's probe: {replies:?}"); + }; + assert_eq!(*target, Some(REJOINING_REPLICA)); + assert!(suffix.is_empty()); + let response = Message::<StartViewHeader>::new(size_of::<StartViewHeader>()) + .transmute_header(|_, header: &mut StartViewHeader| { + header.command = Command::StartView; + header.cluster = 1; + header.replica = NEXT_PRIMARY; + header.view = *view; + header.op = *op; + header.commit = *commit; + header.incarnation = *incarnation; + header.group = *group; + header.size = u32::try_from(size_of::<StartViewHeader>()).unwrap(); + header.seal(); + }); + backup.handle_start_view(plane, response.header(), &[]); + assert_eq!(backup.status(), Status::Normal); + assert_eq!(backup.view(), primary.view()); + assert_eq!(backup.log_view(), primary.log_view()); + assert_eq!(backup.sequencer().current_sequence(), PRIMARY_HEAD); + assert_eq!(backup.commit_max(), LOCAL_HEAD); + assert_eq!(backup.commit_min(), LOCAL_HEAD - 1); + } + // A genuine current primary (no newer view seen) keeps heartbeating. #[test] fn current_primary_keeps_heartbeating() { diff --git a/core/consensus/src/plane_helpers.rs b/core/consensus/src/plane_helpers.rs index 53ac8de05..cb9605c35 100644 --- a/core/consensus/src/plane_helpers.rs +++ b/core/consensus/src/plane_helpers.rs @@ -129,7 +129,7 @@ where header.op <= consensus.commit_min() } -/// Shared chain-replication forwarding to the next replica. +/// Shared chain-replication forwarding, skipping disconnected replicas. /// /// Borrows the message, makes a deep copy for the wire, and lets the caller /// retain ownership for journal append. @@ -137,8 +137,8 @@ where /// # Errors /// /// Returns an error if the prepare cannot be routed or the bus cannot deliver -/// it to the next replica. -/// Callers decide error policy (VSR retransmits from WAL via prepare timeout). +/// it to a connected replica before the end of the chain. +/// Other transport errors remain covered by VSR prepare retransmission. #[allow(clippy::future_not_send)] pub async fn replicate_to_next_in_chain<B, P>( consensus: &VsrConsensus<B, P>, @@ -152,20 +152,22 @@ where return Ok(()); }; let frozen = message.deep_copy().into_generic().into_frozen(); - consensus - .message_bus() - .send_to_replica(next, frozen) - .await - .map_err(Into::into) + forward_to_connected_replica( + consensus, + frozen, + next, + consensus.primary_index(message.header().view), + ) + .await } /// Forward an already validated frozen prepare to the next replica without -/// copying its payload. +/// copying its payload, skipping disconnected replicas. /// /// # Errors /// /// Returns an error if the frame is malformed, cannot be routed, or the bus -/// cannot deliver it to the next replica. +/// cannot deliver it to a connected replica before the end of the chain. #[allow(clippy::future_not_send)] pub async fn replicate_frozen_to_next_in_chain<B, P>( consensus: &VsrConsensus<B, P>, @@ -179,11 +181,42 @@ where let Some(next) = replication_target(consensus, &header)? else { return Ok(()); }; - consensus - .message_bus() - .send_to_replica(next, message) - .await - .map_err(Into::into) + forward_to_connected_replica( + consensus, + message, + next, + consensus.primary_index(header.view), + ) + .await +} + +#[allow(clippy::future_not_send)] +async fn forward_to_connected_replica<B, P>( + consensus: &VsrConsensus<B, P>, + message: Frozen<MESSAGE_ALIGN>, + mut next: u8, + primary: u8, +) -> Result<(), ChainReplicationError> +where + B: MessageBus, + P: Pipeline<Entry = PipelineEntry>, +{ + loop { + match consensus + .message_bus() + .send_to_replica(next, message.clone()) + .await + { + Ok(()) => return Ok(()), + Err(error @ (SendError::ReplicaNotConnected(_) | SendError::ConnectionClosed)) => { + next = (next + 1) % consensus.replica_count(); + if next == primary { + return Err(error.into()); + } + } + Err(error) => return Err(error.into()), + } + } } fn frozen_prepare_header( @@ -905,7 +938,8 @@ pub fn repaired_frontier_update( /// consensus is sans-io and cannot consult the journal itself, so the plane /// that owns the journal must vouch that this exact prepare is durable before /// the ack leaves. `false` withholds the ack; the primary's retransmit -/// re-drives it once a later persist succeeds. +/// re-drives it once a later persist succeeds. Returns `true` only after the +/// acknowledgment is queued for delivery. /// /// # Panics /// - If `header.command` is not `Command::Prepare`. @@ -915,22 +949,23 @@ pub async fn send_prepare_ok<B, P>( consensus: &VsrConsensus<B, P>, header: &PrepareHeader, is_persisted: bool, -) where +) -> bool +where B: MessageBus, P: Pipeline<Entry = PipelineEntry>, { assert_eq!(header.command, Command::Prepare); if consensus.status() != Status::Normal { - return; + return false; } if consensus.is_transferring() { - return; + return false; } if !is_persisted { - return; + return false; } assert!( @@ -941,7 +976,7 @@ pub async fn send_prepare_ok<B, P>( ); if header.op > consensus.sequencer().current_sequence() { - return; + return false; } let prepare_ok_header = PrepareOkHeader { @@ -972,7 +1007,7 @@ pub async fn send_prepare_ok<B, P>( consensus .send_or_loopback(primary, message.into_generic()) - .await; + .await } #[cfg(test)] @@ -1272,6 +1307,118 @@ mod tests { )); } + #[test] + fn given_disconnected_chain_peers_when_forwarding_should_reach_the_next_peer_immediately() { + for frozen in [false, true] { + for (replica, count, view, disconnected, expected) in [ + (0, 3, 0, vec![1], 2), + (1, 3, 1, vec![2], 0), + (2, 5, 1, vec![3, 4], 0), + ] { + let consensus = + VsrConsensus::new(1, replica, count, 0, SpyBus::new(), LocalPipeline::new()); + consensus.init(); + let bus = consensus.message_bus(); + for peer in &disconnected { + bus.failures + .borrow_mut() + .insert(*peer, SendError::ReplicaNotConnected(*peer)); + } + let message = prepare_message(1, 0, 42).transmute_header( + |old, header: &mut PrepareHeader| { + *header = old; + header.view = view; + }, + ); + let result = if frozen { + let prepare = message.deep_copy().into_frozen(); + let pointer = prepare.as_slice().as_ptr(); + let result = futures::executor::block_on(replicate_frozen_to_next_in_chain( + &consensus, prepare, + )); + if let Some((_, sent)) = bus.sent.borrow().first() { + assert_eq!( + sent.as_slice().as_ptr(), + pointer, + "forwarding must not copy the payload" + ); + } + result + } else { + futures::executor::block_on(replicate_to_next_in_chain(&consensus, &message)) + }; + result.expect("a disconnected peer must not delay forwarding to the live suffix"); + assert_eq!( + bus.attempts.borrow().as_slice(), + [disconnected, vec![expected]].concat() + ); + let sent = bus.sent.borrow(); + assert_eq!(sent.len(), 1); + assert_eq!(sent[0].0, expected); + assert_eq!(sent[0].1.as_slice(), message.as_slice()); + assert_eq!( + consensus.commit_max(), + 0, + "forwarding is not an acknowledgement" + ); + } + } + } + + #[test] + fn given_a_chain_boundary_when_forwarding_should_never_wrap_to_the_primary() { + for (replica, count, connected, expected_attempts) in [ + (0, 3, true, vec![1]), + (0, 3, false, vec![1, 2]), + (1, 3, false, vec![2]), + (2, 3, false, vec![]), + (0, 1, false, vec![]), + ] { + let consensus = + VsrConsensus::new(1, replica, count, 0, SpyBus::new(), LocalPipeline::new()); + consensus.init(); + consensus.message_bus().reject_sends.set(!connected); + let result = futures::executor::block_on(replicate_frozen_to_next_in_chain( + &consensus, + prepare_message(1, 0, 42).into_frozen(), + )); + assert_eq!(result.is_ok(), connected || expected_attempts.is_empty()); + assert_eq!( + *consensus.message_bus().attempts.borrow(), + expected_attempts + ); + } + } + + #[test] + fn given_a_transport_error_when_forwarding_should_skip_only_disconnected_peers() { + for error in [ + SendError::ConnectionClosed, + SendError::Backpressure, + SendError::ReplicaRouteMissing(1), + SendError::ReplicaForwardFailed(1), + SendError::BusShuttingDown, + ] { + let skip = matches!(error, SendError::ConnectionClosed); + let consensus = VsrConsensus::new(1, 0, 3, 0, SpyBus::new(), LocalPipeline::new()); + consensus.init(); + consensus + .message_bus() + .failures + .borrow_mut() + .insert(1, error); + let result = futures::executor::block_on(replicate_frozen_to_next_in_chain( + &consensus, + prepare_message(1, 0, 42).into_frozen(), + )); + assert_eq!(result.is_ok(), skip); + assert_eq!( + consensus.message_bus().attempts.borrow().as_slice(), + if skip { &[1, 2][..] } else { &[1][..] }, + ); + } + } + #[test] fn given_committed_prepare_when_selecting_replication_target_should_reject() { let consensus = VsrConsensus::new(1, 0, 3, 0, NoopBus, LocalPipeline::new()); @@ -2028,14 +2175,43 @@ mod tests { ); } + #[test] + fn send_prepare_ok_reports_transport_failure_and_can_be_retried() { + let consensus = VsrConsensus::new(1, 1, 3, 0, SpyBus::new(), LocalPipeline::new()); + consensus.init(); + let header = PrepareHeader { + command: Command::Prepare, + cluster: 1, + checksum: 42, + ..Default::default() + }; + consensus.message_bus().reject_sends.set(true); + assert!(!futures::executor::block_on(send_prepare_ok( + &consensus, &header, true, + ))); + assert!(consensus.message_bus().sent.borrow().is_empty()); + + consensus.message_bus().reject_sends.set(false); + assert!(futures::executor::block_on(send_prepare_ok( + &consensus, &header, true, + ))); + assert_eq!(consensus.message_bus().sent.borrow().len(), 1); + } + struct SpyBus { sent: std::cell::RefCell<Vec<(u8, Frozen<MESSAGE_ALIGN>)>>, + reject_sends: std::cell::Cell<bool>, + failures: std::cell::RefCell<BTreeMap<u8, SendError>>, + attempts: std::cell::RefCell<Vec<u8>>, } impl SpyBus { fn new() -> Self { Self { sent: std::cell::RefCell::new(Vec::new()), + reject_sends: std::cell::Cell::new(false), + failures: std::cell::RefCell::new(BTreeMap::new()), + attempts: std::cell::RefCell::new(Vec::new()), } } } @@ -2056,6 +2232,13 @@ mod tests { replica: u8, data: Frozen<MESSAGE_ALIGN>, ) -> Result<(), SendError> { + self.attempts.borrow_mut().push(replica); + if self.reject_sends.get() { + return Err(SendError::ReplicaNotConnected(replica)); + } + if let Some(error) = self.failures.borrow_mut().remove(&replica) { + return Err(error); + } self.sent.borrow_mut().push((replica, data)); Ok(()) } diff --git a/core/integration/src/harness/disk.rs b/core/integration/src/harness/disk.rs index 4b72890f7..50425d925 100644 --- a/core/integration/src/harness/disk.rs +++ b/core/integration/src/harness/disk.rs @@ -75,9 +75,14 @@ pub fn consumer_offset_file_ids( /// /// Matches the segment file NAME shape, not the `.log` extension alone and not /// a `streams/` path prefix: the server's own text log sits under the same data -/// root, so an extension-only match would count tracing output as segment data. +/// root. The parent must be a partition ID so quarantined copies are excluded. pub fn is_segment_log(path: &Path) -> bool { - path.extension().is_some_and(|extension| extension == "log") + // Quarantine directories and private WAL links are not live partition data. + path.parent() + .and_then(Path::file_name) + .and_then(|name| name.to_str()) + .is_some_and(|name| name.parse::<u32>().is_ok()) + && path.extension().is_some_and(|extension| extension == "log") && path .file_stem() .and_then(|stem| stem.to_str()) @@ -177,7 +182,7 @@ pub fn installed_payloads_complete(data_path: &Path, expected: &[String]) -> Res /// commit cadence, which differs between primary (one flush per op) and backup /// (one flush per committed heartbeat range). fn is_comparable(rel: &str, include_wal: bool) -> bool { - let is_segment = rel.starts_with("streams/") && rel.ends_with(".log"); + let is_segment = rel.starts_with("streams/") && is_segment_log(Path::new(rel)); let is_metadata_wal = rel == "metadata/journal.wal"; is_segment || (include_wal && is_metadata_wal) } @@ -356,7 +361,7 @@ fn total_log_bytes(root: &Path) -> u64 { && let Ok(rel) = path.strip_prefix(root) { let rel = rel.to_string_lossy().replace('\\', "/"); - if rel.starts_with("streams/") && rel.ends_with(".log") { + if is_comparable(&rel, false) { total += fs::metadata(&path).map(|meta| meta.len()).unwrap_or(0); } } diff --git a/core/integration/tests/cluster/crash_recovery_corruption.rs b/core/integration/tests/cluster/crash_recovery_corruption.rs index f22c6859e..fd60deee2 100644 --- a/core/integration/tests/cluster/crash_recovery_corruption.rs +++ b/core/integration/tests/cluster/crash_recovery_corruption.rs @@ -25,8 +25,8 @@ //! - Interior damage to the metadata WAL or a superblock slot can only be //! bit-rot or operator error, never a torn append, so boot must refuse //! loudly and the node heals by rejoining from a clean slate. -//! - Losing a persisted segment body named by the durable WAL also requires -//! refusal and a clean-slate rejoin; rebuilding an index cannot restore it. +//! - Losing a persisted body named by the durable WAL quarantines the partition +//! for peer recovery while preserving the damaged files for inspection. use std::fs; use std::path::{Path, PathBuf}; @@ -76,7 +76,7 @@ const INDEX_AHEAD_BATCHES: u32 = 30; /// real surviving prefix behind the one entry it strands past the log end. const MIN_INDEX_ENTRIES: usize = 4; /// Infix of the directory the refusal path renames a partition's segment files -/// into (`partitions::state_transfer::quarantine_segment_files`). +/// into (`partitions::state_transfer::quarantine_partition_files`). const FENCED_DIR_MARKER: &str = ".fenced."; /// Boot log line recovery emits when the log cannot back the last entry of an /// index (`server::segment_recovery::recover_segment_bounds`): the positive @@ -84,8 +84,7 @@ const FENCED_DIR_MARKER: &str = ".fenced."; /// Distinct from the line the self-contradicting-index check emits, which ends /// "rebuilding it from the log". const INDEX_REBUILD_MARKER: &str = "discarding the index and rebuilding it from a byte-0 walk"; -const PARTITION_WAL_REFUSAL_MARKER: &str = - "cannot recover partition prepare WAL before segment recovery"; +const PARTITION_WAL_REFUSAL_MARKER: &str = "prepare WAL at"; async fn create_stream_and_topic(client: &IggyClient, durability: Durability) { client @@ -574,8 +573,8 @@ async fn given_a_torn_index_tail_when_a_node_recovers_should_not_misalign_subseq /// Replicated storage rebuilds a stale index from the surviving log and refills /// the tail from peers. Persisted storage syncs bodies before publishing WAL -/// references, so deleting that tail loses durable bytes and must refuse boot. -/// A clean-slate state transfer must restore every acknowledged message. +/// references, so deleting that tail must quarantine the damaged partition. +/// Peer recovery must restore every acknowledged message without wiping the node. #[iggy_harness(cluster_nodes = 3)] #[test_matrix([Durability::Replicated, Durability::Persisted])] async fn given_an_index_ahead_of_a_truncated_log_when_a_node_recovers_should_preserve_acked_messages( @@ -641,20 +640,26 @@ async fn given_an_index_ahead_of_a_truncated_log_when_a_node_recovers_should_pre ); if durability == Durability::Persisted { - let error = harness + harness .restart_node(backup) - .expect_err("boot must refuse a persisted body missing from the durable WAL"); + .expect("boot must quarantine the damaged partition for peer recovery"); if stderr_is_captured() { - let diagnostics = error.to_string(); assert!( - diagnostics.contains(PARTITION_WAL_REFUSAL_MARKER), - "boot must refuse the damaged partition WAL, got: {diagnostics}" + harness + .node(backup) + .stdout_contains(PARTITION_WAL_REFUSAL_MARKER), + "boot must diagnose the damaged partition WAL" ); } + let fenced = fenced_segment_paths(&backup_data); + let refused_log = fenced + .iter() + .find(|path| path.file_name() == log_path.file_name()) + .expect("the damaged public segment must be quarantined"); assert_eq!( - fs::read(&log_path).expect("read the refused segment"), + fs::read(refused_log).expect("read the quarantined segment"), truncated, - "refusal must preserve the damaged segment for diagnosis" + "quarantine must preserve the damaged segment for diagnosis" ); let survivors: Vec<usize> = (0..harness.cluster_size()) .filter(|node| *node != backup) @@ -665,9 +670,6 @@ async fn given_an_index_ahead_of_a_truncated_log_when_a_node_recovers_should_pre .unwrap_or_else(|state| { panic!("the surviving quorum must preserve every ack: {state}") }); - harness - .restart_node_from_clean_slate(backup) - .expect("a clean-slate rejoin must restore the missing persisted bodies"); } else { harness.restart_node(backup).unwrap_or_else(|error| { panic!( diff --git a/core/journal/Cargo.toml b/core/journal/Cargo.toml index ba1f33ff9..01443d492 100644 --- a/core/journal/Cargo.toml +++ b/core/journal/Cargo.toml @@ -41,6 +41,9 @@ twox-hash = { workspace = true } [dev-dependencies] tempfile = { workspace = true } +[target.'cfg(target_os = "linux")'.dev-dependencies] +nix = { workspace = true } + [lints.clippy] enum_glob_use = "deny" pedantic = "deny" diff --git a/core/journal/src/durable_storage.rs b/core/journal/src/durable_storage.rs index e526c2050..b76b4b690 100644 --- a/core/journal/src/durable_storage.rs +++ b/core/journal/src/durable_storage.rs @@ -32,6 +32,8 @@ pub enum OpenMode { Read, ReadWrite, Create, + /// Create a file if absent, preserving the inode and bytes if it exists. + CreateOrOpen, } pub struct StorageEntry { @@ -190,8 +192,8 @@ impl DurableStorage for DiskStorage { async fn open(&self, path: &Path, mode: OpenMode) -> io::Result<File> { let mut options = OpenOptions::new(); options.read(true).write(mode != OpenMode::Read); - if mode == OpenMode::Create { - options.create(true).truncate(true); + if matches!(mode, OpenMode::Create | OpenMode::CreateOrOpen) { + options.create(true).truncate(mode == OpenMode::Create); } options.open(path).await } diff --git a/core/journal/src/partition_journal.rs b/core/journal/src/partition_journal.rs index 5c93fddf1..76d5b8223 100644 --- a/core/journal/src/partition_journal.rs +++ b/core/journal/src/partition_journal.rs @@ -1506,6 +1506,21 @@ mod tests { #[cfg(target_os = "linux")] const FILE_BLOCK_BYTES: u64 = 512; + #[cfg(target_os = "linux")] + fn supports_preallocation(directory: &Path, length: u64) -> bool { + let probe = tempfile::tempfile_in(directory).unwrap(); + match nix::fcntl::fallocate( + &probe, + nix::fcntl::FallocateFlags::FALLOC_FL_KEEP_SIZE, + 0, + i64::try_from(length).unwrap(), + ) { + Ok(()) => true, + Err(nix::errno::Errno::EOPNOTSUPP | nix::errno::Errno::ENOSYS) => false, + Err(error) => panic!("preallocation probe failed: {error}"), + } + } + #[compio::test] async fn referenced_bodies_survive_retention_and_checkpoint_without_wal_copies() { let partition = tempdir().unwrap(); @@ -2396,6 +2411,91 @@ mod tests { assert_eq!(journal.prepares().await.unwrap().len(), 1); } + #[compio::test] + async fn malformed_durable_segment_boundaries_are_refused_on_reopen() { + for invalid in [ + "zero size", + "tail generation", + "checkpoint generation", + "tail ordering", + "empty tail bytes", + "empty tail offsets", + "checkpoint ordering", + "empty checkpoint bytes", + "empty checkpoint offsets", + "rewound tail", + ] { + let partition = tempdir().unwrap(); + let directory = partition.path().join("prepares-7"); + let mut journal = PartitionPrepareJournal::open(&directory, 42, 7) + .await + .unwrap(); + journal + .enable_segment_storage(SegmentPosition::default(), PARTITION_WAL_BLOCK_SIZE as u64) + .await + .unwrap(); + let mut state = journal.state; + let segments = state.segment_storage.as_mut().unwrap(); + match invalid { + "zero size" => segments.max_size = 0, + "tail generation" => segments.tail.generation = segments.next_generation, + "checkpoint generation" => { + segments.checkpoint.generation = segments.next_generation; + } + "tail ordering" => segments.tail.position.start_offset = 1, + "empty tail bytes" => segments.tail.position.next_offset = 1, + "empty tail offsets" => segments.tail.position.length = 1, + "checkpoint ordering" => segments.checkpoint.position.start_offset = 1, + "empty checkpoint bytes" => segments.checkpoint.position.next_offset = 1, + "empty checkpoint offsets" => segments.checkpoint.position.length = 1, + "rewound tail" => { + segments.checkpoint.position = SegmentPosition { + length: 1, + next_offset: 1, + ..Default::default() + } + } + _ => unreachable!(), + } + let encoded = state.encode(); + assert!(JournalState::decode(&encoded).is_err(), "{invalid}"); + std::fs::write(directory.join("frontier"), encoded).unwrap(); + drop(journal); + let error = PartitionPrepareJournal::open(&directory, 42, 7) + .await + .err() + .expect(invalid); + assert_eq!( + error.kind(), + io::ErrorKind::InvalidData, + "{invalid}: {error}" + ); + } + } + + #[compio::test] + async fn a_segment_candidate_below_the_checkpoint_cannot_rewind_its_boundary() { + const BODY_BYTES: usize = 4096; + let partition = tempdir().unwrap(); + let directory = partition.path().join("prepares-7"); + let mut journal = PartitionPrepareJournal::open(&directory, 42, 7) + .await + .unwrap(); + journal + .enable_segment_storage(SegmentPosition::default(), (4 * BODY_BYTES) as u64) + .await + .unwrap(); + let first = segment_prepare(1, 0, 0, BODY_BYTES); + let second = segment_prepare(2, first.header().checksum, 1, BODY_BYTES); + journal.append(first.into_frozen()).await.unwrap(); + journal.append(second.into_frozen()).await.unwrap(); + journal.checkpoint(1).await.unwrap(); + let checkpoint = journal.state.segment_storage.unwrap().checkpoint; + assert_eq!(journal.segment_boundary(2).unwrap().position.next_offset, 2); + journal.entries.get_mut(&2).unwrap().next_offset = Some(0); + assert_eq!(journal.segment_boundary(2), Some(checkpoint)); + } + #[compio::test] async fn oversized_durable_segment_layout_is_refused_before_recovery_allocation() { let partition = tempdir().unwrap(); @@ -2571,6 +2671,7 @@ mod tests { for preallocate in [false, true] { let partition = tempdir().unwrap(); + let preallocation_supported = supports_preallocation(partition.path(), SEGMENT_BYTES); let directory = partition.path().join("prepares-7"); let mut journal = PartitionPrepareJournal::open_with_storage_and_capacity( &directory, @@ -2612,11 +2713,13 @@ mod tests { .unwrap(); let metadata = std::fs::metadata(&active).unwrap(); assert_eq!(metadata.len(), BODY_BYTES as u64); - assert_eq!( - metadata.blocks() * FILE_BLOCK_BYTES >= SEGMENT_BYTES, - preallocate, - "recovery must preserve the reservation after trimming unpublished bytes" - ); + if preallocation_supported { + assert_eq!( + metadata.blocks() * FILE_BLOCK_BYTES >= SEGMENT_BYTES, + preallocate, + "recovery must preserve the reservation after trimming unpublished bytes" + ); + } assert_eq!( std::fs::read(&active).unwrap(), second.as_slice()[size_of::<PrepareHeader>()..] @@ -2633,11 +2736,13 @@ mod tests { .unwrap(); let metadata = std::fs::metadata(&active).unwrap(); assert_eq!(metadata.len(), (2 * BODY_BYTES) as u64); - assert_eq!( - metadata.blocks() * FILE_BLOCK_BYTES >= SEGMENT_BYTES, - preallocate, - "replacement segment must follow the preallocation policy" - ); + if preallocation_supported { + assert_eq!( + metadata.blocks() * FILE_BLOCK_BYTES >= SEGMENT_BYTES, + preallocate, + "replacement segment must follow the preallocation policy" + ); + } assert_eq!( std::fs::read(&active).unwrap(), replacement.as_slice()[size_of::<PrepareHeader>()..] @@ -2704,7 +2809,9 @@ mod tests { ) .unwrap(); assert_eq!(installed.len(), initial.length); - assert!(installed.blocks() * FILE_BLOCK_BYTES >= (4 * BODY_BYTES) as u64); + if supports_preallocation(partition.path(), (4 * BODY_BYTES) as u64) { + assert!(installed.blocks() * FILE_BLOCK_BYTES >= (4 * BODY_BYTES) as u64); + } } let checkpoint_reference = journal.segment_reference(checkpoint.header()).unwrap(); if !materialized { @@ -2783,6 +2890,7 @@ mod tests { #[compio::test] async fn owned_segments_migrate_only_unmaterialized_legacy_prepares() { + const EXISTING_GENERATION: u64 = 41; const BODY_BYTES: usize = 8192; let partition = tempdir().unwrap(); let directory = partition.path().join("prepares-7"); @@ -2791,9 +2899,17 @@ mod tests { .unwrap(); let first = segment_prepare(1, 0, 0, BODY_BYTES); let second = segment_prepare(2, first.header().checksum, 1, BODY_BYTES); - journal.append(first.clone().into_frozen()).await.unwrap(); + let existing_reference = + write_segment(partition.path(), EXISTING_GENERATION, 0, &first).await; + journal + .append_batch_referenced_buffered( + &[first.clone().into_frozen()], + &[Some(existing_reference)], + ) + .await + .unwrap(); + journal.sync().await.unwrap(); journal.append(second.clone().into_frozen()).await.unwrap(); - write_segment(partition.path(), 0, 0, &first).await; let checkpoint = SegmentPosition { start_offset: 0, length: BODY_BYTES as u64, @@ -2803,8 +2919,17 @@ mod tests { .enable_segment_storage(checkpoint, (4 * BODY_BYTES) as u64) .await .unwrap(); - assert!(journal.segment_reference(first.header()).is_none()); - assert!(journal.segment_reference(second.header()).is_some()); + assert_eq!( + journal.segment_reference(first.header()), + Some(existing_reference) + ); + assert!( + journal + .segment_reference(second.header()) + .unwrap() + .generation + > EXISTING_GENERATION + ); drop(journal); let journal = PartitionPrepareJournal::open(&directory, 42, 7) .await diff --git a/core/journal/src/partition_journal/segments.rs b/core/journal/src/partition_journal/segments.rs index a454500ca..6ff31cbb8 100644 --- a/core/journal/src/partition_journal/segments.rs +++ b/core/journal/src/partition_journal/segments.rs @@ -64,7 +64,8 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> { /// The initial boundary must describe durable materialized messages only. /// /// # Errors - /// Returns an error for inconsistent boundaries or a failed storage barrier. + /// Returns an error for inconsistent boundaries, a `max_size` differing from an + /// already enabled layout, or a failed storage barrier. pub async fn enable_segment_storage( &mut self, initial: SegmentPosition, @@ -163,6 +164,7 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> { } /// Install a transferred checkpoint after all replacement segment files are durable. + /// The replacement establishes its own segment size; it does not extend the old layout. /// /// # Errors /// Returns an error if the checkpoint contradicts the installed bytes or a barrier fails. @@ -516,7 +518,17 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> { .ok_or_else(|| invalid("segment rollback handle is absent"))?; let length = file.length().await?; if length < cursor.position.length { - return Err(invalid("segment lost durably published bytes")); + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "segment {} (generation {}, start offset {}) lost durably published bytes: expected at least {}, found {}", + cursor.path(&self.directory).display(), + cursor.generation, + cursor.position.start_offset, + cursor.position.length, + length, + ), + )); } if length > cursor.position.length { file.truncate(cursor.position.length).await?; @@ -611,25 +623,32 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> { let public = parent.join(format!("{start_offset:020}.log")); let file = if self.storage.exists(&retained).await? { self.storage.open(&retained, OpenMode::ReadWrite).await? - } else if length > 0 - || (self.storage.exists(&public).await? + } else { + if length == 0 + && self.storage.exists(&public).await? && self .storage .open(&public, OpenMode::Read) .await? .length() .await? - == 0) - { + > 0 + { + // A purged inode can still be retained by older prepares. + self.remove_segment_name(&public).await?; + } + // Segment roll can create the public name during any await. Both + // creators must open that inode without truncation, then retain it. + let mode = if length == 0 { + OpenMode::CreateOrOpen + } else { + OpenMode::ReadWrite + }; + let file = self.storage.open(&public, mode).await?; self.storage.hard_link(&public, &retained).await?; - self.storage.open(&retained, OpenMode::ReadWrite).await? - } else { - let file = self.storage.open(&retained, OpenMode::Create).await?; - // Offset names can still point at a purged inode retained by older - // prepares. Unlink before reuse; never truncate that inode in place. - self.remove_segment_name(&public).await?; - self.storage.hard_link(&retained, &public).await?; - if let Some(size) = preallocate_size { + if length == 0 + && let Some(size) = preallocate_size + { file.preallocate(&retained, size); } file diff --git a/core/metadata/src/stm/stream.rs b/core/metadata/src/stm/stream.rs index 2582fd420..9844d552d 100644 --- a/core/metadata/src/stm/stream.rs +++ b/core/metadata/src/stm/stream.rs @@ -2801,7 +2801,9 @@ mod tests { stream_id: WireIdentifier::numeric(0), partitions_count: 1, name: WireName::new("t").unwrap(), - options: explicit.to_wire().unwrap(), + options: explicit + .to_explicit_wire(|key| key == topic_option_keys::MESSAGE_EXPIRY) + .unwrap(), }, derived_options: derived.to_wire().unwrap(), partitions: vec![CreatedPartitionAssignment { @@ -2830,7 +2832,10 @@ mod tests { ] { let key = HeaderKey::from_str(policy).unwrap(); let option = topic.options.get(&key).unwrap(); - assert!(option.explicit); + assert!( + !option.explicit, + "unsupplied durability comes from admission defaults" + ); assert_eq!(option.value.kind(), HeaderKind::String); assert_eq!(option.value.as_str().unwrap(), "replicated"); } diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 9436183f5..fdd8e3b64 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -947,6 +947,9 @@ where } pub async fn checkpoint_persistence(&mut self, config: &PartitionsConfig) { + if self.fatal.is_some() { + return; + } let Some(persistence) = self .persistence .as_ref() @@ -1066,6 +1069,9 @@ where } pub async fn drive_persistence(&mut self) { + if self.fatal.is_some() { + return; + } let Some(persistence) = self.persistence.as_ref() else { return; }; @@ -2700,9 +2706,11 @@ where ) -> Result<(u64, bool), IggyError> { if self.persistence.is_some() { let result = crate::offset_storage::read_offset_max(path, offset).await?; - self.write_consumer_offset(path, result.offset, false) - .await?; - Ok((result.offset, true)) + if result.written { + self.write_consumer_offset(path, result.offset, false) + .await?; + } + Ok((result.offset, result.written)) } else { let result = persist_offset_max(path, offset, persisted).await?; Ok((result.offset, result.written)) @@ -2713,7 +2721,8 @@ where &self, pending: PendingConsumerOffsetCommit, ) -> Result<(), IggyError> { - // The replicated WAL already protects these cursor updates until checkpoint. + // For either offset policy, the WAL protects these updates until checkpoint + // syncs their retained writers and directories before reclaiming history. let persisted = self.consumer_offset_durability().is_persisted() && self.persistence.is_none(); let path = self.persisted_offset_path(pending.kind, pending.consumer_id); @@ -8027,7 +8036,7 @@ where } async fn send_prepare_ok(&self, header: &PrepareHeader) -> bool { - if self.materialization_missing { + if self.fatal.is_some() || self.materialization_missing { return false; } // Durable-before-send: a PrepareOk implies this replica's @@ -8062,8 +8071,7 @@ where // that reaches here is journal-backed and ACKs as durable. // (`header_by_op` is a linear scan, so re-proving that here would // put O(journal) on every ack; the call-order invariant stands in.) - send_prepare_ok_common(self.consensus(), header, true).await; - true + send_prepare_ok_common(self.consensus(), header, true).await } } @@ -8988,6 +8996,17 @@ mod tests { for preallocate in [false, true] { let directory = tempfile::tempdir().unwrap(); + let probe = tempfile::tempfile_in(directory.path()).unwrap(); + let preallocation_supported = match nix::fcntl::fallocate( + &probe, + nix::fcntl::FallocateFlags::FALLOC_FL_KEEP_SIZE, + 0, + i64::try_from(SEGMENT_BYTES).unwrap(), + ) { + Ok(()) => true, + Err(nix::errno::Errno::EOPNOTSUPP | nix::errno::Errno::ENOSYS) => false, + Err(error) => panic!("preallocation probe failed: {error}"), + }; let mut partition = partition_at_view(0, 0); partition.set_partition_dir(directory.path().to_string_lossy().into_owned()); partition.runtime_options.durability = iggy_common::Durability::Persisted; @@ -9002,11 +9021,13 @@ mod tests { let empty = std::fs::metadata(directory.path().join("00000000000000000000.log")).unwrap(); assert_eq!(empty.len(), 0); - assert_eq!( - empty.blocks() * BLOCK_BYTES >= SEGMENT_BYTES, - preallocate, - "empty active segment allocation must follow preallocate_segments={preallocate}" - ); + if preallocation_supported { + assert_eq!( + empty.blocks() * BLOCK_BYTES >= SEGMENT_BYTES, + preallocate, + "empty active segment allocation must follow preallocate_segments={preallocate}" + ); + } let persistence = Rc::clone(partition.persistence.as_ref().unwrap()); let namespace = partition.namespace(); let bodies = [ @@ -9043,11 +9064,13 @@ mod tests { let metadata = std::fs::metadata(&rotated).unwrap(); assert_eq!(std::fs::read(&rotated).unwrap(), bodies[1]); assert_eq!(metadata.len(), bodies[1].len() as u64); - assert_eq!( - metadata.blocks() * BLOCK_BYTES >= SEGMENT_BYTES, - preallocate, - "rotated segment allocation must follow preallocate_segments={preallocate}" - ); + if preallocation_supported { + assert_eq!( + metadata.blocks() * BLOCK_BYTES >= SEGMENT_BYTES, + preallocate, + "rotated segment allocation must follow preallocate_segments={preallocate}" + ); + } assert_eq!(partition.log.active_segment().size.as_bytes_u64(), 0); } } @@ -9240,8 +9263,11 @@ mod tests { assert_eq!(partition.mint_frontier(), 1); } - #[compio::test] - async fn deferred_purge_preserves_a_durable_primary_self_ack_until_it_can_be_sent() { + async fn partition_with_pending_durable_ack() -> ( + tempfile::TempDir, + IggyPartition<RecordingBus>, + PrepareHeader, + ) { let directory = tempfile::tempdir().unwrap(); let (mut partition, _) = recording_partition_at(0, 3); partition.set_partition_dir(directory.path().to_string_lossy().into_owned()); @@ -9266,6 +9292,83 @@ mod tests { partition.start_persistence(); persistence.drain_with_timeout().await.unwrap(); + (directory, partition, header) + } + + #[compio::test] + async fn persisted_ack_survives_recovery_transfer_and_a_rewound_head() { + let (_directory, mut partition, header) = partition_with_pending_durable_ack().await; + partition.consensus().begin_view_probe(); + partition.drive_persistence().await; + assert_eq!(partition.pending_persisted_acks.borrow().len(), 1); + + partition.consensus().init(); + partition.consensus().begin_state_transfer_await(); + partition.drive_persistence().await; + assert_eq!(partition.pending_persisted_acks.borrow().len(), 1); + + partition + .consensus() + .set_state_transfer_stage(consensus::StateTransferStage::Idle); + partition + .consensus() + .sequencer() + .set_sequence(header.op - 1); + partition.drive_persistence().await; + assert_eq!(partition.pending_persisted_acks.borrow().len(), 1); + let mut acknowledgments = Vec::new(); + partition + .consensus() + .drain_loopback_into(&mut acknowledgments); + assert!(acknowledgments.is_empty()); + + partition.consensus().sequencer().set_sequence(header.op); + partition.drive_persistence().await; + partition.drive_persistence().await; + partition + .consensus() + .drain_loopback_into(&mut acknowledgments); + assert!(partition.pending_persisted_acks.borrow().is_empty()); + assert_eq!(acknowledgments.len(), 1); + let ack = bytemuck::checked::from_bytes::<PrepareOkHeader>(acknowledgments[0].as_slice()); + assert_eq!(ack.op, header.op); + assert_eq!(ack.prepare_checksum, header.checksum); + } + + #[compio::test] + async fn persisted_ack_remains_fenced_after_a_local_commit_failure() { + let (_directory, mut partition, header) = partition_with_pending_durable_ack().await; + partition.fatal = Some(FatalCommit { + namespace_raw: partition.namespace().inner(), + op: header.op, + operation: Operation::StoreConsumerOffset, + }); + partition.persistence.as_ref().unwrap().request_checkpoint(); + partition + .consensus() + .restore_commit_state(header.op, header.op); + partition.checkpoint_persistence(&repair_config()).await; + partition.drive_persistence().await; + partition.acknowledge_prepare(header.op).await; + let mut acknowledgments = Vec::new(); + partition + .consensus() + .drain_loopback_into(&mut acknowledgments); + assert!( + acknowledgments.is_empty(), + "a fenced partition must never acknowledge" + ); + assert_eq!(partition.pending_persisted_acks.borrow().len(), 1); + assert_eq!( + partition.fatal().unwrap().operation, + Operation::StoreConsumerOffset + ); + } + + #[compio::test] + async fn deferred_purge_preserves_a_durable_primary_self_ack_until_it_can_be_sent() { + let (_directory, mut partition, header) = partition_with_pending_durable_ack().await; + partition.purge_deferred = true; let mut acknowledgments = Vec::new(); for _ in 0..2 { @@ -12024,6 +12127,50 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + #[compio::test] + async fn wal_backed_cold_offsets_skip_covered_values_and_write_advances() { + let (directory, partition, _) = partition_with_pending_durable_ack().await; + let path = directory.path().join("offset"); + let path = path.to_str().unwrap(); + persist_offset(path, 114, false).await.unwrap(); + assert_eq!( + partition + .write_cold_consumer_offset(path, 109, true) + .await + .unwrap(), + (114, false) + ); + assert_eq!( + partition + .write_cold_consumer_offset(path, 114, true) + .await + .unwrap(), + (114, false) + ); + assert!( + partition + .persistence + .as_ref() + .unwrap() + .take_offset_file(path) + .is_none() + ); + assert_eq!( + partition + .write_cold_consumer_offset(path, 115, true) + .await + .unwrap(), + (115, true) + ); + assert_eq!( + crate::offset_storage::read_offset_max(path, 0) + .await + .unwrap() + .offset, + 115 + ); + } + /// The persisted-offset tracker is cold after a restart; the first /// auto-commit folds against the file once (so a pre-existing higher value /// wins, exactly like the old per-commit read-modify-write) and warms the diff --git a/core/partitions/src/offset_storage.rs b/core/partitions/src/offset_storage.rs index 180b4bd6f..aaf341412 100644 --- a/core/partitions/src/offset_storage.rs +++ b/core/partitions/src/offset_storage.rs @@ -133,6 +133,10 @@ pub async fn persist_offset(path: &str, offset: u64, persisted: bool) -> Result< /// Return the write result with its original descriptor so checkpoint observes /// writeback errors even after an unsuccessful write. /// +/// No barrier runs here. For either offset durability policy, the caller must +/// retain the writer and sync it and its directory before reclaiming the WAL +/// history that protects the update. +/// /// # Errors /// The outer error reports directory/open failures before writing begins. pub async fn persist_offset_retained( diff --git a/core/partitions/src/persistence.rs b/core/partitions/src/persistence.rs index cddc7fb78..427574707 100644 --- a/core/partitions/src/persistence.rs +++ b/core/partitions/src/persistence.rs @@ -727,9 +727,6 @@ impl<S: DurableStorage> PartitionPersistence<S> { /// /// # Errors /// Returns a barrier error that the caller must fence like a failed write. - /// - /// # Panics - /// Panics if the previous writer was not taken before replacement. pub async fn retain_offset_file(&self, path: String, file: S::File) -> io::Result<()> { let retained_count = self.offset_files.borrow().len() + self.retired_offset_files.borrow().len(); @@ -739,18 +736,15 @@ impl<S: DurableStorage> PartitionPersistence<S> { let Some(permit) = OffsetFilePermit::acquire() else { return file.sync().await; }; - assert!( - self.offset_files - .borrow_mut() - .insert( - path, - RetainedOffsetFile { - file, - _permit: permit - } - ) - .is_none() - ); + if let Some(previous) = self.offset_files.borrow_mut().insert( + path, + RetainedOffsetFile { + file, + _permit: permit, + }, + ) { + self.retired_offset_files.borrow_mut().push(previous); + } Ok(()) } diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs index bb1787917..af1ed37bb 100644 --- a/core/partitions/src/state_transfer.rs +++ b/core/partitions/src/state_transfer.rs @@ -1630,6 +1630,9 @@ async fn clear_materialization_missing(directory: &str) -> std::io::Result<()> { /// Move every segment file in `partition_dir` aside into `<dir>.fenced.<n>/`, /// returning the directory used. /// +/// Boot recovery also supplies `wal_revision` to +/// move the refused prepare WAL. Live callers leave it unset to retain open writers. +/// /// The partition directory itself STAYS, and so do its two superblock slots: /// they hold the group's only durable `(view, log_view)`, and moving them would /// make the rebuild read an empty directory -- no `restore_partition_view`, @@ -1648,7 +1651,10 @@ async fn clear_materialization_missing(directory: &str) -> std::io::Result<()> { /// the rebuild plants segment 0 with `file_exists = false` and truncates /// whatever the failed quarantine left, so callers tombstone the partition and /// leave the bytes for an operator. -pub async fn quarantine_segment_files(partition_dir: &str) -> std::io::Result<String> { +pub async fn quarantine_partition_files( + partition_dir: &str, + wal_revision: Option<u64>, +) -> std::io::Result<String> { // `create_dir`, not stat-then-create: one syscall per attempt instead of // two, and race-free. Deliberately NOT `create_dir_all`, which succeeds on // an existing directory and would silently merge this fence into an earlier @@ -1686,6 +1692,19 @@ pub async fn quarantine_segment_files(partition_dir: &str) -> std::io::Result<St }; compio::fs::rename(&path, &PathBuf::from(&target).join(name)).await?; } + if let Some(revision) = wal_revision { + let name = format!("prepares-{revision}"); + match compio::fs::rename( + &Path::new(partition_dir).join(&name), + &Path::new(&target).join(&name), + ) + .await + { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(error), + } + } // All three touched directories: the target (its new dirents), the source // (the removals), and the source's parent (the target directory itself is a // new dirent there). Without the target-side syncs a crash can leave the @@ -2185,13 +2204,13 @@ where Ok(checksum) } - /// [`quarantine_segment_files`] over this partition's directory, for the + /// [`quarantine_partition_files`] over this partition's directory, for the /// shard's `ConvergeFailed` fence -- the safety argument (segment files /// move, superblock slots STAY, copies are unreclaimed operator evidence) /// lives on the free function. `None` for an in-memory partition. /// /// # Errors - /// The underlying `std::io::Error`; see [`quarantine_segment_files`] for why + /// The underlying `std::io::Error`; see [`quarantine_partition_files`] for why /// a failure is not something the rebuild can absorb. pub async fn quarantine_partition_dir(&self) -> std::io::Result<Option<String>> { let Some(dir) = self.partition_dir.clone() else { @@ -2204,7 +2223,7 @@ where if self.consensus().replica_count() > 1 { mark_materialization_missing(&dir, self.created_revision).await?; } - quarantine_segment_files(&dir).await.map(Some) + quarantine_partition_files(&dir, None).await.map(Some) } /// Release the cached offer once no requester holds one (the shard's @@ -3483,9 +3502,6 @@ where minted_next_offset: u64, staged_was_empty: bool, ) -> Result<(), iggy_common::IggyError> { - if let Some(persistence) = &self.persistence { - persistence.retire_offset_files(); - } // The empty plant below can land on a base offset this sweep unlinks, // so an in-flight poll's cached read fd would keep serving the retired // inodes as live data. Same hazard and same fix as `purge`. diff --git a/core/server/config.toml b/core/server/config.toml index 78f01111d..9d51eb0a8 100644 --- a/core/server/config.toml +++ b/core/server/config.toml @@ -849,14 +849,16 @@ clients_table_max = 8192 # ("unlimited" when unset) # message_expiry - delete sealed segments older than this ("none" when unset) # Both policies can be active at once. The active segment is never touched. +# WAL-backed partitions request a checkpoint before deleting referenced segments; +# retention waits for that checkpoint, even below the usual WAL capacity trigger. # Call GET /options/topic (or the SDK's describe_options) for the full catalog # with this server's defaults. # Topic durability and flush scheduling are set at creation and returned by GetTopic. # durability controls message completion. consumer_offset_durability controls explicit offset stores and deletes. # Both independently default to "replicated". Neither inherits the other. -# In replicated groups, persisted messages use WAL references to segment bodies, retained by hard link until reclamation. -# Persisted offsets with replicated messages use inline WAL bodies. Both layouts charge full message payloads to wal_bytes_max. +# In replicated groups, either persisted policy enables WAL references to segment bodies, retained by hard link until reclamation. +# This includes replicated messages with persisted offsets. Full message payloads count against wal_bytes_max. # Their barriers can also persist co-batched messages. Only replicated/replicated stays off this WAL. # Persisted prepares are limited to 64 MiB including the consensus header. Oversized operations are rejected permanently. # Both policies write data to storage. "replicated" waits for VSR commit and application without an additional stable-storage barrier. diff --git a/core/server/src/dispatch/partition.rs b/core/server/src/dispatch/partition.rs index 4ee599786..de74599e0 100644 --- a/core/server/src/dispatch/partition.rs +++ b/core/server/src/dispatch/partition.rs @@ -1409,6 +1409,8 @@ mod tests { use crate::dispatch::test_support::{ SpyBus, TestMux, TestShard, prepare_message, request_message, test_shard, }; + #[cfg(target_os = "linux")] + use consensus::Sequencer; use iggy_binary_protocol::ReplyHeader; use iggy_binary_protocol::primitives::partition_assignment::CreatedPartitionAssignment; use iggy_binary_protocol::requests::consumer_offsets::DeleteConsumerOffsetRequest; @@ -1432,6 +1434,112 @@ mod tests { ShardIdentity, shard_channel, }; + #[cfg(target_os = "linux")] + #[compio::test] + async fn checkpoint_index_failure_is_returned_by_the_same_partition_tick() { + let root = tempfile::tempdir().unwrap(); + let bus = SpyBus::default(); + let shard = test_shard(&bus, 0, 3, 1); + let namespace = IggyNamespace::new(1, 1, 0); + let consensus = consensus::VsrConsensus::new( + 1, + 0, + 3, + namespace.inner(), + bus, + consensus::LocalPipeline::new(), + ); + consensus.init(); + let mut partition = partitions::IggyPartition::with_in_memory_storage( + std::sync::Arc::new(iggy_common::PartitionStats::default()), + consensus, + shard.plane.partitions().config().segment_size, + ); + partition.set_runtime_options(iggy_common::TopicRuntimeOptions { + durability: iggy_common::Durability::Persisted, + preallocate_segments: Some(false), + ..Default::default() + }); + partition.set_partition_dir(root.path().to_string_lossy().into_owned()); + let capacity = journal::partition_journal::PARTITION_WAL_BYTES_MAX; + let (persistence, prepares) = partitions::PartitionPersistence::open_with_capacity( + &root.path().join("prepares-0"), + namespace.inner(), + 0, + journal::durable_storage::DiskStorage, + capacity, + false, + ) + .await + .unwrap(); + partition + .open_persistence_with_recovered(capacity, Some((Rc::clone(&persistence), prepares))) + .await + .unwrap(); + + let mut messages = server_common::send_messages::IggyMessages::with_capacity(1); + messages.push(server_common::send_messages::IggyMessage { + header: server_common::send_messages::IggyMessageHeader::default(), + payload: Bytes::from_static(b"checkpoint"), + user_headers: None, + }); + let batch = + server_common::send_messages::SendMessagesOwned::from_messages(namespace, &messages) + .unwrap(); + let mut body = vec![0; batch.header.total_size()]; + batch.header.encode_into(&mut body); + body[iggy_binary_protocol::batch::BATCH_HEADER_SIZE..].copy_from_slice(&batch.blob); + let prepare = prepare_message(Operation::SendMessages, 1, 1, &body).transmute_header( + |original, header: &mut PrepareHeader| { + *header = original; + header.cluster = 1; + header.group = namespace.inner(); + header.checksum = header.identity_checksum(); + }, + ); + let checksum = prepare.header().checksum; + persistence + .append(prepare.clone().into_frozen(), true) + .unwrap(); + assert!(persistence.start()); + Rc::clone(&persistence).run().await; + partition + .log + .journal() + .inner + .append(prepare.into_frozen()) + .await + .unwrap(); + partition.log.journal_mut().info.messages_count = 1; + partition.log.journal_mut().info.size = iggy_common::IggyByteSize::from(body.len() as u64); + partition.consensus().sequencer().set_sequence(1); + partition.consensus().set_last_prepare_checksum(checksum); + partition.consensus().restore_commit_state(1, 1); + partition.log.index_writers_mut()[0] = Some(Rc::new( + partitions::IggyIndexWriter::new( + "/dev/full", + Rc::new(std::sync::atomic::AtomicU64::new(0)), + false, + false, + ) + .await + .unwrap(), + )); + persistence.request_checkpoint(); + assert!(partition.needs_persistence_checkpoint()); + assert!(partition.fatal().is_none()); + shard.plane.partitions().insert(namespace, partition); + + let fault = shard + .tick_partitions(&mut Vec::new()) + .await + .expect("the checkpoint fault must be returned in its originating sweep"); + assert_eq!(fault.namespace_raw, namespace.inner()); + assert_eq!(fault.op, 1); + assert_eq!(fault.operation, Operation::SendMessages); + assert_eq!(persistence.checkpoint_op(), 0); + } + #[compio::test] async fn given_invalid_partition_writes_when_resolving_should_preserve_offset_error_codes() { const VSR_CLIENT: u128 = 1; diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs index 3366b5952..cebe04091 100644 --- a/core/server/src/partition_helpers.rs +++ b/core/server/src/partition_helpers.rs @@ -701,7 +701,12 @@ pub async fn load_partition_or_fence( ServerError::Iggy(Box::new(IggyError::CannotSyncFile)) })?; } - match partitions::state_transfer::quarantine_segment_files(&partition_dir).await { + match partitions::state_transfer::quarantine_partition_files( + &partition_dir, + (replica_count > 1).then_some(partition_metadata.created_revision), + ) + .await + { Ok(fenced_dir) => error!( stream_id, topic_id, @@ -877,9 +882,20 @@ async fn load_partition( .unwrap_or(iggy_common::DEFAULT_PREALLOCATE_SEGMENTS), ) .await - .map_err(|error| { - warn!(%error, "cannot recover partition prepare WAL before segment recovery"); - ServerError::from(IggyError::CannotReadFile) + .map_err(|source| match source.kind() { + std::io::ErrorKind::InvalidData | std::io::ErrorKind::UnexpectedEof => { + ServerError::PartitionRecoveryRefused { + dir: PathBuf::from(&partition_dir), + stream_id: namespace.stream_id(), + topic_id: namespace.topic_id(), + partition_id: namespace.partition_id(), + reason: PartitionRecoveryRefusal::PrepareWal { directory, source }, + } + } + _ => ServerError::PartitionPrepareWalIo { + dir: directory, + source, + }, })?, ) } else { @@ -1698,6 +1714,75 @@ mod tests { ); } + #[compio::test] + async fn corrupt_prepare_wal_is_quarantined_without_losing_the_recovery_fence() { + let root = tempfile::tempdir().unwrap(); + let config = solo_config(&root); + let namespace = IggyNamespace::new(1, 1, 0); + let runtime = TopicRuntimeOptions { + durability: iggy_common::Durability::Persisted, + preallocate_segments: Some(false), + ..Default::default() + }; + drop( + build_partition_fresh( + &config, + namespace, + Arc::new(PartitionStats::default()), + 0, + runtime, + CLUSTER, + REPLICA, + REPLICAS, + 0, + Rc::new(IggyMessageBus::new(0)), + ) + .await + .unwrap(), + ); + let directory = config.get_partition_path(1, 1, 0); + let (store, _) = open_partition_superblock(&directory, test_identity()) + .await + .unwrap(); + let state = recorded_state(3, 2); + store.write(&state.to_bytes()).await.unwrap(); + drop(store); + let frontier = Path::new(&directory).join("prepares-0/frontier"); + let mut corrupt = std::fs::read(&frontier).unwrap(); + corrupt[0] ^= u8::MAX; + std::fs::write(&frontier, &corrupt).unwrap(); + let partitions = solo_partitions(); + let metadata = Partition::new(0, namespace.inner(), IggyTimestamp::now(), 0, 0); + + for _ in 0..2 { + let partition = load_partition_or_fence( + &config, + namespace, + Arc::new(PartitionStats::default()), + &metadata, + runtime, + CLUSTER, + REPLICA, + REPLICAS, + Rc::new(IggyMessageBus::new(0)), + &partitions, + ) + .await + .unwrap() + .unwrap(); + assert!(partition.requires_state_transfer()); + assert!(partition.consensus().view() >= state.view); + let (_, recovered) = open_partition_superblock(&directory, test_identity()) + .await + .unwrap(); + assert_eq!(recovered, Some(state)); + } + assert_eq!( + std::fs::read(format!("{directory}.fenced.0/prepares-0/frontier")).unwrap(), + corrupt, + ); + } + fn recorded_state(view: u32, log_view: u32) -> VsrState { VsrState { cluster: CLUSTER, diff --git a/core/server/src/responses.rs b/core/server/src/responses.rs index 9f4df1739..c5fca28a9 100644 --- a/core/server/src/responses.rs +++ b/core/server/src/responses.rs @@ -1072,7 +1072,7 @@ fn topic_option_descriptors() -> Result<Vec<OptionDescriptor>, IggyError> { key: WireName::new(topic_option_keys::CONSUMER_OFFSET_DURABILITY).map_err(|_| IggyError::InvalidFormat)?, kind: HeaderKind::String.as_code(), default_value: Bytes::from_static(b"replicated"), - description: "Explicit offset completion: replicated or persisted. Independently defaults to replicated. Poll auto-commit remains asynchronous. In replicated groups, persisted offsets also journal full message bodies, even with replicated message durability, and spend partition.wal_bytes_max on them.".to_string(), + description: "Explicit offset completion: replicated or persisted. Independently defaults to replicated. Poll auto-commit remains asynchronous. In replicated groups, persisted offsets also enable WAL references to segment bodies, retaining their inodes by hard link until reclamation, even with replicated message durability. Full body sizes count against partition.wal_bytes_max.".to_string(), }, OptionDescriptor { key: WireName::new(topic_option_keys::MESSAGES_REQUIRED_TO_SAVE) diff --git a/core/server/src/segment_recovery.rs b/core/server/src/segment_recovery.rs index ff8ed5560..ca82b42bf 100644 --- a/core/server/src/segment_recovery.rs +++ b/core/server/src/segment_recovery.rs @@ -286,9 +286,9 @@ pub async fn load_persisted_segments_with_checkpoint( && bounds.as_ref().map_or(0, |bounds| bounds.messages_size) != messages_size { return Err( - identity.refusal(PartitionRecoveryRefusal::StorageSizeMismatch { + identity.refusal(PartitionRecoveryRefusal::CheckpointSizeMismatch { start_offset, - on_disk_bytes: bounds.as_ref().map_or(0, |bounds| bounds.messages_size), + validated_bytes: bounds.as_ref().map_or(0, |bounds| bounds.messages_size), expected_bytes: messages_size, }), ); diff --git a/core/server/src/server_error.rs b/core/server/src/server_error.rs index 775c0345c..b32a8ab7f 100644 --- a/core/server/src/server_error.rs +++ b/core/server/src/server_error.rs @@ -161,6 +161,12 @@ pub enum ServerError { #[source] source: std::io::Error, }, + #[error("failed to recover partition prepare WAL at {dir}: {source}")] + PartitionPrepareWalIo { + dir: PathBuf, + #[source] + source: std::io::Error, + }, // Quarantines the one partition rather than treating the group as fresh or // reading through to a superseded view: mirrors the metadata plane's // `RecoveryError::SuperblockUnreadable` policy, minus the boot refusal, @@ -212,7 +218,7 @@ pub enum ServerError { // catch this error, and only they log it -- a claim here would render // beside theirs and contradict one branch or the other. #[error( - "partition {stream_id}/{topic_id}/{partition_id} at {dir} refused segment \ + "partition {stream_id}/{topic_id}/{partition_id} at {dir} refused storage \ recovery: {reason}" )] PartitionRecoveryRefused { @@ -466,8 +472,16 @@ pub enum PartitionRecoveryRefusal { walked_position: u64, durable_position: u64, }, - /// A writer reopening over recovered bounds found the on-disk length - /// diverging from the size recovery just validated and truncated to. + PrepareWal { + directory: PathBuf, + source: std::io::Error, + }, + CheckpointSizeMismatch { + start_offset: u64, + validated_bytes: u64, + expected_bytes: u64, + }, + /// The physical file length differs from the required recovered boundary. StorageSizeMismatch { start_offset: u64, on_disk_bytes: u64, @@ -579,6 +593,20 @@ impl std::fmt::Display for PartitionRecoveryRefusal { the log has lost previously durable bytes mid-chunk, so rebuilding \ would re-mint their offsets" ), + Self::PrepareWal { directory, source } => write!( + f, + "prepare WAL at {} cannot be recovered: {source}", + directory.display() + ), + Self::CheckpointSizeMismatch { + start_offset, + validated_bytes, + expected_bytes, + } => write!( + f, + "segment {start_offset} validated prefix has {validated_bytes} bytes, \ + but the WAL checkpoint requires {expected_bytes}" + ), Self::StorageSizeMismatch { start_offset, on_disk_bytes, @@ -586,7 +614,7 @@ impl std::fmt::Display for PartitionRecoveryRefusal { } => write!( f, "segment {start_offset} file length {on_disk_bytes} diverged from \ - its recovered size {expected_bytes} at writer open" + its required recovered size {expected_bytes}" ), } } diff --git a/core/server_common/src/fs_utils.rs b/core/server_common/src/fs_utils.rs index 883c9b28f..c64090c56 100644 --- a/core/server_common/src/fs_utils.rs +++ b/core/server_common/src/fs_utils.rs @@ -65,23 +65,8 @@ pub fn preallocate_file(file: &fs::File, file_path: &Path, len: u64) { return; }; - // Runs INLINE on the shard thread, deliberately. `server_common::executor` - // sets `thread_pool_limit(0)` on the shard proactor, so `spawn_blocking` - // has no worker to park a task on and compio panics the shard outright with - // "the thread pool is needed but no worker thread is running". (That limit - // is skipped on macOS, whose polling driver routes fs through the pool, so - // the panic is Linux-and-most-targets, not universal. This arm is - // Linux-only regardless.) - // - // The cost is acceptable only because of what this call is: a metadata-only - // extent reservation, microseconds on the local filesystems this option - // exists for, and an immediate `EOPNOTSUPP` where the filesystem cannot do - // it. Where it can genuinely block -- NFSv4.2 `ALLOCATE`, FUSE, a badly - // fragmented extent tree forcing a journal commit -- it stalls the whole - // core, not one partition, because nothing here yields. Preallocation is - // opt-in per topic at creation for that reason; on such a deployment, - // create topics without `preallocate_segments` rather than reintroducing a - // pool the shard runtime does not have. + // Shard runtimes disable the worker pool, so this opt-in reservation runs + // inline. Slow filesystem allocation stalls the shard until it returns. if let Err(error) = fallocate(file, FallocateFlags::FALLOC_FL_KEEP_SIZE, 0, len) { warn!( target: "iggy.partitions.storage", diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 3d94ec366..adf172f24 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -7608,6 +7608,12 @@ where walk_cursor.get_or_insert(namespace); } } + if let Some(fault) = partition.fatal() { + if fatal.is_none() { + fatal = Some(fault.clone()); + } + continue; + } let consensus_view = partition.consensus().view(); let commit_min = partition.consensus().commit_min(); let cluster = partition.consensus().cluster(); diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs index 5dd28930e..01569de1b 100644 --- a/core/shard/src/metrics.rs +++ b/core/shard/src/metrics.rs @@ -34,7 +34,7 @@ //! compio reactor contexts. Each shard owns its own instance, and the server //! exposes every shard's instance through the `[http.metrics]` scrape //! endpoint via [`ShardMetrics::register`] (one `shard`-labelled -//! sub-registry per shard); every drop site also logs via `tracing`. +//! sub-registry per shard). Drop-site tracing supplements the counters. use prometheus_client::encoding::EncodeLabelSet; use prometheus_client::metrics::counter::Counter; @@ -113,6 +113,7 @@ pub mod frame_drop_variant { /// status and no frame of the client's was dropped, so counting it with /// shed frames would read as a routing loss. pub const PARTITION_AUTO_COMMIT: &str = "partition_auto_commit"; + pub const PARTITION_PERSISTENCE_COMPLETED: &str = "partition_persistence_completed"; } /// Reason labels used in `frame_drops_total`. @@ -161,7 +162,7 @@ pub mod frame_drop_reason { // pair enters the `Family` (and therefore the scrape) the first time a drop // site actually produces it, so the unreachable corners of the 7 x 9 cross // product never appear as permanent zero-valued series. -const VARIANT_COUNT: usize = 8; +const VARIANT_COUNT: usize = 9; const REASON_COUNT: usize = 11; const VARIANTS: [&str; VARIANT_COUNT] = [ @@ -173,6 +174,7 @@ const VARIANTS: [&str; VARIANT_COUNT] = [ frame_drop_variant::METADATA_COMMIT_TICK, frame_drop_variant::REPLICA_HANDSHAKE_ACK, frame_drop_variant::PARTITION_AUTO_COMMIT, + frame_drop_variant::PARTITION_PERSISTENCE_COMPLETED, ]; const REASONS: [&str; REASON_COUNT] = [ diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs index 1f653a48b..0eea8a678 100644 --- a/core/shard/src/router.rs +++ b/core/shard/src/router.rs @@ -287,12 +287,17 @@ where M: RestorableMetadataStm, { if let Some(sender) = self.senders.get(self.id as usize).cloned() { + let metrics = self.metrics.clone(); self.plane .partitions() .set_persistence_notifier(Rc::new(move |completion| { - let _ = sender.try_send(ShardFrame::lifecycle( - LifecycleFrame::PartitionPersistenceCompleted(completion), - )); + let frame = LifecycleFrame::PartitionPersistenceCompleted(completion); + if let Err(error) = sender.try_send(ShardFrame::lifecycle(frame)) { + metrics.record_frame_drop( + frame_drop_variant::PARTITION_PERSISTENCE_COMPLETED, + crate::coordinator::classify_try_send_err(&error), + ); + } })); } // Reused across every pump iteration; pre-size to skip the diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs index 0ec322cc6..a828bbe80 100644 --- a/core/simulator/src/replica.rs +++ b/core/simulator/src/replica.rs @@ -381,7 +381,7 @@ pub fn new_shard( size_of_messages_required_to_save: IggyByteSize::from(4 * 1024 * 1024), validate_checksum: true, - segment_size: IggyByteSize::from(1024 * 1024 * 1024), + segment_size: IggyByteSize::from(iggy_common::DEFAULT_SEGMENT_SIZE), preallocate_segments: false, encryptor: None, path_layout: PartitionPathLayout::default(), diff --git a/core/simulator/src/storage.rs b/core/simulator/src/storage.rs index a2d860865..d8100dc09 100644 --- a/core/simulator/src/storage.rs +++ b/core/simulator/src/storage.rs @@ -206,17 +206,23 @@ impl DurableStorage for SimStorage { type File = SimFile; async fn open(&self, path: &Path, mode: OpenMode) -> io::Result<SimFile> { - let operation = if mode == OpenMode::Create { + let creates = matches!(mode, OpenMode::Create | OpenMode::CreateOrOpen); + let operation = if creates { StorageOperation::Create } else { StorageOperation::Open }; + self.wait_for(operation).await; let (inode, epoch) = self.perform(operation, |state, _| { - let inode = if mode == OpenMode::Create { + let inode = if creates { let (parent, name) = state.parent(path)?; if let Some(&inode) = state.directory(parent)?.get(&name) { match &mut state.inodes[inode] { - Inode::File { buffered, .. } => buffered.clear(), + Inode::File { buffered, .. } => { + if mode == OpenMode::Create { + buffered.clear(); + } + } Inode::Directory { .. } => { return Err(invalid("cannot truncate directory")); } @@ -316,6 +322,7 @@ impl DurableStorage for SimStorage { } async fn hard_link(&self, source: &Path, target: &Path) -> io::Result<()> { + self.wait_for(StorageOperation::Link).await; self.perform(StorageOperation::Link, |state, _| { let inode = state.lookup(source)?; let (parent, name) = state.parent(target)?; diff --git a/core/simulator/src/storage/tests.rs b/core/simulator/src/storage/tests.rs index d2496bd99..190140171 100644 --- a/core/simulator/src/storage/tests.rs +++ b/core/simulator/src/storage/tests.rs @@ -616,6 +616,46 @@ fn metadata_only_append_skips_segment_barriers_after_durable_bodies() { }); } +#[test] +fn replacing_a_retained_offset_writer_keeps_both_inodes_until_checkpoint() { + block_on(async { + let (storage, persistence) = queued_batch(1).await; + assert!(persistence.start()); + Rc::clone(&persistence).run().await; + let path = Path::new("/partition/offset"); + let original = Path::new("/partition/original-offset"); + let mut file = storage.open(path, OpenMode::Create).await.unwrap(); + file.write(0, b"original".to_vec()).await.unwrap(); + storage.hard_link(path, original).await.unwrap(); + persistence + .retain_offset_file(path.to_str().unwrap().to_owned(), file) + .await + .unwrap(); + storage.remove_file(path).await.unwrap(); + let mut replacement = storage.open(path, OpenMode::Create).await.unwrap(); + replacement.write(0, b"replaced".to_vec()).await.unwrap(); + persistence + .retain_offset_file(path.to_str().unwrap().to_owned(), replacement) + .await + .unwrap(); + + persistence.checkpoint_files( + 1, + vec![path.to_path_buf()], + vec![Path::new(DIRECTORY).to_path_buf()], + ); + assert!(persistence.start()); + Rc::clone(&persistence).run().await; + assert!(persistence.failure().is_none()); + assert_eq!(persistence.checkpoint_op(), 1); + storage.crash(Crash::PowerLoss); + let file = storage.open(original, OpenMode::Read).await.unwrap(); + assert_eq!(file.read(0, 8).await.unwrap(), b"original"); + let file = storage.open(path, OpenMode::Read).await.unwrap(); + assert_eq!(file.read(0, 8).await.unwrap(), b"replaced"); + }); +} + #[test] fn a_full_offset_writer_cache_synchronizes_overflow_and_reports_barrier_failure() { const OFFSET_KEYS: usize = 128; @@ -1260,6 +1300,74 @@ async fn baseline() -> (SimStorage, PartitionPrepareJournal<SimStorage>) { (storage, journal) } +#[test] +fn segment_roll_during_wal_create_keeps_the_same_inode() { + block_on(segment_roll_during_wal_open(StorageOperation::Create)); +} + +#[test] +fn segment_roll_during_wal_link_keeps_the_same_inode() { + block_on(segment_roll_during_wal_open(StorageOperation::Link)); +} + +async fn segment_roll_during_wal_open(operation: StorageOperation) { + let storage = storage_for_partition().await; + let mut journal = + PartitionPrepareJournal::open_with_storage(Path::new(WAL), 42, 7, storage.clone()) + .await + .unwrap(); + journal + .enable_segment_storage(SegmentPosition::default(), OWNED_BATCH_BYTES as u64) + .await + .unwrap(); + let first = owned_prepare(1, 0, 0); + journal.append(first.clone().into_frozen()).await.unwrap(); + let second = owned_prepare(2, first.header().checksum, 1); + let public = Path::new(DIRECTORY).join(format!("{:020}.log", 1)); + let retained = Path::new(WAL).join("segment-1-1.log"); + storage.state.borrow_mut().paused = Some(operation); + let mut append = Box::pin(journal.append(second.clone().into_frozen())); + assert!(poll!(&mut append).is_pending()); + storage.resume(); + let roll_reader = storage.open(&public, OpenMode::CreateOrOpen).await.unwrap(); + append + .await + .expect("a concurrent segment roll must not fail the WAL append"); + { + let state = storage.state.borrow(); + assert_eq!( + state.lookup(&public).unwrap(), + state.lookup(&retained).unwrap() + ); + assert_eq!( + roll_reader.inode, + state.lookup(&public).unwrap(), + "the WAL must retain the inode opened by the segment roll" + ); + } + assert_eq!( + roll_reader.read(0, OWNED_BATCH_BYTES).await.unwrap(), + second.as_slice()[size_of::<PrepareHeader>()..] + ); + drop(journal); + storage.crash(Crash::PowerLoss); + let recovered = + PartitionPrepareJournal::open_with_storage(Path::new(WAL), 42, 7, storage.clone()) + .await + .unwrap(); + assert_eq!(recovered.durable_op(), 2); + let actual = recovered.prepares().await.unwrap(); + for (actual, expected) in actual.iter().zip([first, second]) { + assert_eq!(actual.as_slice(), expected.as_slice()); + } + assert_eq!(actual.len(), 2); + let state = storage.state.borrow(); + assert_eq!( + state.lookup(&public).unwrap(), + state.lookup(&retained).unwrap() + ); +} + #[test] fn buffered_owned_segments_rotate_without_barriers_and_persist_offset_predecessors() { block_on(async {
