This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch fix_journal_panic in repository https://gitbox.apache.org/repos/asf/iggy.git
commit b329c960d4c2fd6dfcec64ffab9095ad91599c57 Author: Grzegorz Koszyk <[email protected]> AuthorDate: Wed Oct 7 09:41:34 2026 +0200 review fixes --- core/metadata/src/impls/metadata.rs | 181 ++++++++++++++++++++++++++++++++++-- core/metadata/src/lib.rs | 4 +- core/shard/src/lib.rs | 43 +++++---- core/simulator/src/lib.rs | 119 ++++++++++++++++++++++++ 4 files changed, 318 insertions(+), 29 deletions(-) diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index f334a74d7..f07ce20e5 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -234,6 +234,26 @@ impl IggySnapshot { /// `commit_min < commit_max` every tick. const COMMIT_WALK_OPS_MAX: usize = 64; +/// Why one [`IggyMetadata::commit_journal`] call stopped walking. +/// +/// Only [`Self::BodyMissing`] names an op whose header the journal holds but +/// cannot serve. A [`Self::PipelineHead`] op is committed from the journal by +/// the pipeline driver (`commit_committable_prefix`, via `on_ack` or +/// `resume_stranded_commits`), so its prepare must stay resident. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CommitWalkStop { + /// `commit_min` reached `commit_max`. + CaughtUp, + /// The per-call walk cap was reached; the next call resumes. + WalkCap, + /// The next op is the pipeline head; the pipeline driver commits it. + PipelineHead, + /// No header is resident at `op`. + HeaderMissing { op: u64 }, + /// The header at `op` is resident but the WAL cannot produce its body. + BodyMissing { op: u64 }, +} + /// Integrity framing is mandatory because the superblock can lag snapshot publication. const SNAPSHOT_TRAILER_MAGIC: u32 = 0x4953_4E50; @@ -3963,9 +3983,12 @@ where /// `is_caught_up_primary` on a higher "applied frontier". /// /// `is_caught_up_primary_gate_states` pins clauses, NOT intra-loop window. + /// + /// Returns why the walk stopped; callers that only drive the walk ignore + /// it. #[allow(clippy::cast_possible_truncation, clippy::missing_panics_doc)] #[allow(clippy::future_not_send)] - pub async fn commit_journal(&self) { + pub async fn commit_journal(&self) -> CommitWalkStop { let consensus = self.consensus.as_ref().unwrap(); let journal = self.journal.as_ref().unwrap(); @@ -3976,7 +3999,7 @@ where "commit_journal: stopping at op={} after {applied} ops; resuming next tick", consensus.commit_min() ); - break; + return CommitWalkStop::WalkCap; } applied += 1; let op = consensus.commit_min() + 1; @@ -3996,7 +4019,7 @@ where .pipeline_head_header() .is_some_and(|head| head.op == op) { - break; + return CommitWalkStop::PipelineHead; } let Some(header) = journal.handle().header(op as usize) else { @@ -4007,13 +4030,13 @@ where // StartView adoption, or by `tick_metadata`'s gap detector, // which is the only one of those a live drop under sustained // traffic reaches. - break; + return CommitWalkStop::HeaderMissing { op }; }; let header = *header; let Some(prepare) = journal.handle().entry(&header).await else { warn!("commit_journal: prepare body missing for op={op}, stopping"); - break; + return CommitWalkStop::BodyMissing { op }; }; // SM apply + client_table mutation BEFORE `advance_commit_min` @@ -4030,6 +4053,7 @@ where self.advance_applied_frontier(op); debug!("commit_journal: committed op={op}"); } + CommitWalkStop::CaughtUp } fn observe_prepare_runtime_state(&self, prepare: &Message<PrepareHeader>) { @@ -5902,7 +5926,7 @@ mod tests { // committed by the group and resident here, none of it walked. consensus.advance_commit_max(OPS); - md.commit_journal().await; + assert_eq!(md.commit_journal().await, CommitWalkStop::WalkCap); assert_eq!( consensus.commit_min(), CAP, @@ -5910,7 +5934,7 @@ mod tests { as the resident run is, however long that is" ); - md.commit_journal().await; + assert_eq!(md.commit_journal().await, CommitWalkStop::CaughtUp); assert_eq!( consensus.commit_min(), OPS, @@ -5919,6 +5943,149 @@ mod tests { ); } + /// The walk leaves the pipeline head to `on_ack` and says so. + /// + /// A primary-elect inherits a committed head at `commit_min + 1` and the + /// walk moves nothing over it. `tick_metadata`'s stall detector truncates + /// on `BodyMissing` only, so this has to come back as `PipelineHead` with + /// the prepare still journaled for the pipeline driver to read. + #[compio::test] + async fn given_committed_pipeline_head_at_next_op_when_walking_should_stop_as_pipeline_head() { + const CLIENT: u128 = 1; + const SESSION: u64 = 1; + const ACTING_USER: u32 = 7; + + let dir = tempfile::tempdir().unwrap(); + let journal = + journal::prepare_journal::PrepareJournal::open(&dir.path().join("journal.wal"), 0) + .await + .unwrap(); + let consensus = VsrConsensus::new( + 1, + 0, + 1, + server_common::sharding::METADATA_GROUP, + NoopBus, + LocalPipeline::new(), + ); + consensus.init(); + let md: IggyMetadata<_, journal::prepare_journal::PrepareJournal, (), TestMux> = + IggyMetadata::new( + Some(consensus), + Some(journal), + None, + None, + TestMux::default(), + None, + ); + let consensus = md.consensus.as_ref().unwrap(); + md.client_table + .borrow_mut() + .commit_register( + CLIENT, + ACTING_USER, + [0x5a; 32], + register_reply(CLIENT, SESSION), + ) + .unwrap(); + + let prepare = md + .prepare_request(create_stream_request(CLIENT, 1, "s1")) + .expect("CreateStream is client-allowed"); + consensus.pipeline_message(PlaneKind::Metadata, &prepare); + md.on_replicate(prepare).await; + // Committed with no driver to apply it: the inherited-head shape. + consensus.advance_commit_max(1); + + assert_eq!(md.commit_journal().await, CommitWalkStop::PipelineHead); + assert_eq!( + consensus.commit_min(), + 0, + "the walk applied the pipeline head" + ); + assert!( + md.journal.as_ref().unwrap().handle().header(1).is_some(), + "the head's prepare must stay journaled for the pipeline driver" + ); + + md.resume_stranded_commits().await; + assert_eq!( + consensus.commit_min(), + 1, + "the pipeline driver commits the head the walk left to it" + ); + } + + /// A committed op with no resident header is a plain gap: the walk names it + /// as `HeaderMissing`, never as a missing body that needs truncating. + #[compio::test] + async fn given_committed_op_without_header_when_walking_should_stop_as_header_missing() { + const CLIENT: u128 = 1; + const SESSION: u64 = 1; + const ACTING_USER: u32 = 7; + const RESIDENT_OPS: u64 = 2; + + let dir = tempfile::tempdir().unwrap(); + std::fs::create_dir_all(dir.path().join(crate::impls::METADATA_DIR)).unwrap(); + let journal = + journal::prepare_journal::PrepareJournal::open(&dir.path().join("journal.wal"), 0) + .await + .unwrap(); + // A backup, so no pipeline head caps the walk. + let consensus = VsrConsensus::new( + 1, + 1, + 3, + server_common::sharding::METADATA_GROUP, + NoopBus, + LocalPipeline::new(), + ); + consensus.init(); + let md: IggyMetadata<_, journal::prepare_journal::PrepareJournal, (), TestMux> = + IggyMetadata::new( + Some(consensus), + Some(journal), + None, + None, + TestMux::default(), + Some(dir.path().to_path_buf()), + ); + let consensus = md.consensus.as_ref().unwrap(); + md.client_table + .borrow_mut() + .commit_register( + CLIENT, + ACTING_USER, + [0x5a; 32], + register_reply(CLIENT, SESSION), + ) + .unwrap(); + + for request in 1..=RESIDENT_OPS { + let prepare = md + .prepare_request(create_stream_request( + CLIENT, + request, + &format!("s{request}"), + )) + .expect("CreateStream is client-allowed"); + md.on_replicate(prepare).await; + } + consensus.advance_commit_max(RESIDENT_OPS + 1); + + assert_eq!( + md.commit_journal().await, + CommitWalkStop::HeaderMissing { + op: RESIDENT_OPS + 1 + } + ); + assert_eq!( + consensus.commit_min(), + RESIDENT_OPS, + "the walk applies the resident run before stopping at the gap" + ); + } + /// A state-transfer receiver admits the first live prepare above the floor it /// installed, instead of waiting for an op the snapshot already contains. /// diff --git a/core/metadata/src/lib.rs b/core/metadata/src/lib.rs index bc771641c..c475cc7b2 100644 --- a/core/metadata/src/lib.rs +++ b/core/metadata/src/lib.rs @@ -28,8 +28,8 @@ pub use applied_frontier::{AppliedFrontier, Reached}; // Re-export IggyMetadata for use in other modules pub use impls::metadata::{ - BoundSession, CommitNotifier, IggyMetadata, MetadataSubmitError, StateTransferOffer, - apply_committed_prepare, + BoundSession, CommitNotifier, CommitWalkStop, IggyMetadata, MetadataSubmitError, + StateTransferOffer, apply_committed_prepare, }; // Recovery vocabulary other crates name in their own signatures and error diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index eff6cf5da..2527318a8 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -65,7 +65,7 @@ use message_bus::{BusMessage, ConnectionPermit, MessageBus, SharedTlsServerConfi use metadata::IggyMetadata; use metadata::impls::metadata::StreamsFrontend; use metadata::stm::StateMachine; -use metadata::{BoundSession, MetadataSubmitError}; +use metadata::{BoundSession, CommitWalkStop, MetadataSubmitError}; use partitions::state_transfer::TransferArtifact; use partitions::{ FatalCommit, IggyPartition, IggyPartitions, PollFragments, PollingArgs, PollingConsumer, @@ -1438,10 +1438,10 @@ where /// metadata group per node (precedent: [`Self::metadata_transfer_attempts`]). metadata_gap_ticks: Cell<u32>, - /// Op the tick's commit walk last stopped on without moving, or `0`. The - /// journal names it but cannot produce its body, so the gap probe counts it - /// as absent and lets repair fetch it. Cleared implicitly: any advance of - /// `commit_min` makes it stop matching `commit_min + 1`. + /// Op of the commit walk's last `BodyMissing` stop, or `0`. The gap probe + /// counts it as absent so repair can refetch it. Zeroed by any other + /// walk-arm stop; an advance of `commit_min` past it also stops it + /// matching `commit_min + 1`. metadata_walk_stuck_op: Cell<u64>, /// Serving-side cache of state-transfer offers, both planes, keyed by @@ -10129,8 +10129,9 @@ where /// /// Nothing else clears it: `on_repair_prepare` returns early for an op /// whose header is resident, and the append under it is refused anyway. - /// `stuck_op` is at `commit_min + 1` under `commit_max`, so a quorum holds - /// it and repair can serve it back. + /// `stuck_op` is the op the walk stopped on: `commit_min + 1` once this + /// tick's walk has advanced, and at or under `commit_max`, so a quorum + /// holds it and repair can serve it back. /// /// SERIALIZATION: same argument as `reconcile_metadata_view_divergence`, /// which is the other shard-side `truncate_from` caller. This runs on the @@ -10202,14 +10203,14 @@ where // // The header ring is only half of what the walk needs. `commit_journal` // reads the BODY through `entry()`, which answers `None` for an op the - // ring names but the WAL cannot produce, and then breaks without moving - // `commit_min`. Reading the body here instead is not an option (it is an - // async WAL read, per tick, on the walk's fast path), so the walk - // reports the op it stopped on and this treats that op as absent -- - // which it is, for every purpose this probe serves. Without it the two - // disagree forever: the walk cannot move, the probe keeps calling the - // group walk-stalled, the debounce keeps resetting, and repair never - // arms. + // ring names but the WAL cannot produce, and stops there with + // `CommitWalkStop::BodyMissing`. Reading the body here instead is not an + // option (it is an async WAL read, per tick, on the walk's fast path), + // so the walk reports the op it stopped on and this treats that op as + // absent, which it is for every purpose this probe serves. Without it + // the two disagree forever: the walk cannot move, the probe keeps + // calling the group walk-stalled, the debounce keeps resetting, and + // repair never arms. // // Self-clearing: any path that advances `commit_min` past the stuck op // leaves `stuck_op != commit_min + 1`, so nothing has to retract it. @@ -10375,15 +10376,17 @@ where commit_max = probe.commit_max, "metadata commit walk parked over resident committed ops; resuming" ); - metadata.commit_journal().await; - // A walk that moved nothing found the header and not the body. + // Truncate on `BodyMissing` alone. A walk that moved nothing is + // not proof: `PipelineHead` moves nothing either, and dropping + // that prepare panics the pipeline driver that commits it next. + // // Recording the op stops the detector calling this a parked // walk, but arming repair alone cannot refill it: the ingest // skips an op whose header is resident and `append` refuses the // slot under it, so the header has to go first. - let walked = consensus.commit_min(); - if walked == probe.commit_min { - let stuck_op = walked.saturating_add(1); + if let CommitWalkStop::BodyMissing { op: stuck_op } = + metadata.commit_journal().await + { // Once per op: a failed truncation leaves the header where // it is, and retrying every tick only repeats the error. if self.metadata_walk_stuck_op.replace(stuck_op) != stuck_op { diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index c176dd5d2..16d0644e8 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -8754,3 +8754,122 @@ mod probe_answer_divergence_tests { ); } } + +#[cfg(test)] +mod metadata_walk_stuck_detector_tests { + //! The metadata walk-stall detector in `tick_metadata` must not mistake the + //! pipeline head for a header without a body. + //! + //! `commit_journal` stops, without moving `commit_min`, at the op the + //! pipeline holds, because committing it is `on_ack`'s job. A detector that + //! read "the walk moved nothing" as "the body is unreadable" would truncate + //! the WAL from that op, leaving the primary's pipeline with a committed + //! header whose prepare is gone, and the next commit driver would panic + //! with `committed prepare ... must be in journal`. + + use consensus::Consensus; + + use super::*; + + /// Swarm seed whose network lets replica 2 learn `commit_max = 3` without + /// ever walking op 3, and then elects it primary of view 2. + const SEED: u64 = 1189; + const REPLICAS: u8 = 3; + const CLIENTS: [u128; 3] = [1, 2, 3]; + + const VIEW_0_PRIMARY: u8 = 0; + const VIEW_1_PRIMARY: u8 = 1; + /// The replica the faults leave to win view 2. + const PRIMARY_ELECT: usize = 2; + + /// Replica faults as `(step, replica, crash)`. The view-0 primary goes + /// down and comes back, then the view-1 primary goes down, which puts the + /// view-2 change on the primary-elect. + const FAULTS: [(usize, u8, bool); 3] = [ + (297, VIEW_0_PRIMARY, true), + (705, VIEW_0_PRIMARY, false), + (1_151, VIEW_1_PRIMARY, true), + ]; + + /// Past the view-2 start (step ~1313), then enough to heal and drain. + const FAULT_STEPS: usize = 1_400; + const DRAIN_STEPS: usize = 4_000; + + /// Only the three `Register`s go through the metadata plane: no client + /// traffic, no metadata workload. Op 3 is the last of them. + const REGISTERED_OPS: u64 = 3; + + #[test] + fn given_primary_elect_inheriting_a_committed_head_when_the_tick_probes_the_walk_should_keep_its_prepare() + { + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { + enabled: false, + size: iggy_common::IggyByteSize::from(0u64), + bucket_capacity: 1, + }); + let mut network = packet::PacketSimulatorOptions::swarm(SEED); + network.node_count = REPLICAS; + network.client_count = u8::try_from(CLIENTS.len()).expect("few clients"); + let mut sim = Simulator::new(usize::from(REPLICAS), CLIENTS.iter().copied(), network); + sim.init_partition(IggyNamespace::new(1, 1, 0)); + for id in CLIENTS { + sim.register_client_with_primary(&SimClient::new(id)); + } + + let mut inherited_head_seen = false; + for step in 0..FAULT_STEPS { + for &(at, replica, crash) in &FAULTS { + if at == step { + if crash { + sim.replica_crash(replica); + } else { + sim.replica_restart(replica); + } + } + } + sim.step(); + inherited_head_seen |= primary_elect_holds_inherited_head(&sim); + } + // Without this a drifted seed that never builds the shape passes the + // drain assertion below without testing anything. + assert!( + inherited_head_seen, + "replica {PRIMARY_ELECT} never led view 2 with a committed pipeline head at \ + commit_min + 1; the seed no longer reaches the shape under test" + ); + sim.replica_restart(VIEW_1_PRIMARY); + for _ in 0..DRAIN_STEPS { + sim.step(); + } + + // Every replica applies every registration: a dropped pipeline prepare + // must neither panic the primary nor leave any replica short of it. + for replica in 0..usize::from(REPLICAS) { + let consensus = sim + .metadata_consensus(replica) + .expect("shard 0 hosts the metadata plane"); + assert!( + consensus.commit_min() >= REGISTERED_OPS, + "replica {replica} applied metadata only through op {}, short of the \ + {REGISTERED_OPS} registrations", + consensus.commit_min() + ); + } + } + + /// Whether the primary-elect is a normal view-2 primary whose pipeline head + /// is committed and is the very next op it owes the state machine. Gated + /// like `group_is_walk_stalled`, so a view-change window does not count. + fn primary_elect_holds_inherited_head(sim: &Simulator) -> bool { + let consensus = sim + .metadata_consensus(PRIMARY_ELECT) + .expect("shard 0 hosts the metadata plane"); + consensus.is_primary() + && consensus.is_normal() + && !consensus.is_transferring() + && consensus.view() >= 2 + && consensus.pipeline_head_header().is_some_and(|head| { + head.op == consensus.commit_min() + 1 && head.op <= consensus.commit_max() + }) + } +}
