This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch fix_version_fencing in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 47a015b59f47937647f325393b1b47b6b453feb0 Author: Grzegorz Koszyk <[email protected]> AuthorDate: Thu Aug 20 16:31:35 2026 +0200 fix(shard): tell an unknown consensus operation from a corrupt header --- core/binary_protocol/src/consensus/error.rs | 6 ++ core/binary_protocol/src/consensus/header.rs | 14 +++++ core/binary_protocol/src/consensus/operation.rs | 11 ++++ core/server/src/partition_reconciler.rs | 59 ++++++++++++++++++++ core/server_common/src/consensus_message.rs | 27 +++++++-- core/shard/src/metrics.rs | 12 +++- core/shard/src/router.rs | 73 ++++++++++++++++--------- 7 files changed, 170 insertions(+), 32 deletions(-) diff --git a/core/binary_protocol/src/consensus/error.rs b/core/binary_protocol/src/consensus/error.rs index 02fb044e6..8b6a4ba28 100644 --- a/core/binary_protocol/src/consensus/error.rs +++ b/core/binary_protocol/src/consensus/error.rs @@ -79,6 +79,12 @@ pub enum ConsensusError { #[error("invalid bit pattern in header (enum discriminant out of range)")] InvalidBitPattern, + #[error( + "operation {operation:#04x} is not known to this build; the sender runs a release that \ + added a consensus operation, so this node cannot journal or ack the frame" + )] + UnsupportedOperation { operation: u8 }, + #[error("client-bound command {0:?} cannot be dispatched on inbound path")] ClientBoundCommand(Command), } diff --git a/core/binary_protocol/src/consensus/header.rs b/core/binary_protocol/src/consensus/header.rs index 2f205df3d..8fe0150cc 100644 --- a/core/binary_protocol/src/consensus/header.rs +++ b/core/binary_protocol/src/consensus/header.rs @@ -103,6 +103,15 @@ pub fn frame_checksum_bytes(header: &[u8; HEADER_SIZE]) -> u128 { pub trait ConsensusHeader: Sized + CheckedBitPattern + NoUninit { const COMMAND: Command; + /// Byte offset of this header's `operation` field, `None` when it carries + /// none. + /// + /// The typed decode reads the raw byte here after a failed checked cast, + /// so an operation a newer release added is reported as version skew + /// rather than corruption. An offset rather than a getter because the cast + /// has already failed by then, so no typed view of the header exists. + const OPERATION_OFFSET: Option<usize> = None; + /// Whether a frame carrying `command` may be typed as this header. /// Defaults to an exact match; a header that serves several commands /// with one layout (e.g. `RepairDone` / `RangeEvicted`) widens it. @@ -477,6 +486,7 @@ fn validate_request_fields( } impl ConsensusHeader for RoutedRequestHeader { + const OPERATION_OFFSET: Option<usize> = Some(core::mem::offset_of!(Self, operation)); const COMMAND: Command = Command::Request; /// The client-wire [`RequestHeader`] this is promoted from is unsealed, and the /// promotion copies `checksum` verbatim, so there is nothing here to verify. @@ -511,6 +521,7 @@ impl ConsensusHeader for RoutedRequestHeader { } impl ConsensusHeader for RequestHeader { + const OPERATION_OFFSET: Option<usize> = Some(core::mem::offset_of!(Self, operation)); const COMMAND: Command = Command::Request; const FRAME_SEALED: bool = false; @@ -621,6 +632,7 @@ impl Default for ReplyHeader { } impl ConsensusHeader for ReplyHeader { + const OPERATION_OFFSET: Option<usize> = Some(core::mem::offset_of!(Self, operation)); const COMMAND: Command = Command::Reply; const FRAME_SEALED: bool = false; @@ -971,6 +983,7 @@ impl Default for PrepareHeader { } impl ConsensusHeader for PrepareHeader { + const OPERATION_OFFSET: Option<usize> = Some(core::mem::offset_of!(Self, operation)); const COMMAND: Command = Command::Prepare; const FRAME_SEALED: bool = false; @@ -1178,6 +1191,7 @@ impl Default for PrepareOkHeader { } impl ConsensusHeader for PrepareOkHeader { + const OPERATION_OFFSET: Option<usize> = Some(core::mem::offset_of!(Self, operation)); const FRAME_SEALED: bool = true; const COMMAND: Command = Command::PrepareOk; diff --git a/core/binary_protocol/src/consensus/operation.rs b/core/binary_protocol/src/consensus/operation.rs index 691bd9f1b..95de1d4ee 100644 --- a/core/binary_protocol/src/consensus/operation.rs +++ b/core/binary_protocol/src/consensus/operation.rs @@ -98,6 +98,17 @@ pub enum Operation { } impl Operation { + /// Whether `code` is a discriminant this build defines. + /// + /// The typed decode needs this to tell an operation a newer release added + /// from a corrupted header byte: bytemuck's checked cast rejects both with + /// one undifferentiated error, and only the former is fixable by upgrading + /// this node. + #[must_use] + pub fn is_known_code(code: u8) -> bool { + bytemuck::checked::try_cast::<u8, Self>(code).is_ok() + } + pub const INTERNAL_START: u8 = Self::CreateTopicWithAssignments as u8; pub const METADATA_START: u8 = Self::CreateStream as u8; pub const PARTITION_START: u8 = Self::SendMessages as u8; diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs index 4df3e5402..0ab7d1be8 100644 --- a/core/server/src/partition_reconciler.rs +++ b/core/server/src/partition_reconciler.rs @@ -2350,6 +2350,65 @@ mod tests { ); } + /// The mixed-version upgrade hole from IGGY-250, at the router seam this + /// time: a wire-valid consensus frame carrying an operation only a newer + /// release defines must leave an accounted, operator-visible trace instead + /// of a bare warn log, because the frame's group stops making progress + /// until this node is upgraded. + #[compio::test] + async fn given_an_unknown_operation_when_dispatched_should_account_an_upgrade_fence_drop() { + // Far past every discriminant this build defines. + const OPERATION_FROM_A_NEWER_RELEASE: u8 = 0xEE; + + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let shard = build_test_shard(0, &config, TestMux::default()); + + let mut owned = server_common::iobuf::Owned::<{ server_common::MESSAGE_ALIGN }>::zeroed( + iggy_binary_protocol::HEADER_SIZE, + ); + { + let frame = owned.as_mut_slice(); + let size_offset = std::mem::offset_of!(PrepareHeader, size); + let frame_size = + u32::try_from(iggy_binary_protocol::HEADER_SIZE).expect("header size fits in u32"); + frame[size_offset..size_offset + 4].copy_from_slice(&frame_size.to_le_bytes()); + frame[std::mem::offset_of!(PrepareHeader, command)] = Command::Prepare as u8; + frame[std::mem::offset_of!(PrepareHeader, operation)] = OPERATION_FROM_A_NEWER_RELEASE; + let header: &[u8; iggy_binary_protocol::HEADER_SIZE] = frame + [..iggy_binary_protocol::HEADER_SIZE] + .try_into() + .expect("frame spans a full header"); + let checksum = iggy_binary_protocol::frame_checksum_bytes(header); + frame[..size_of::<u128>()].copy_from_slice(&checksum.to_le_bytes()); + } + let message = Message::<iggy_binary_protocol::GenericHeader>::try_from(owned) + .expect("a sealed Prepare frame is wire-valid in the generic view"); + + let before = shard.metrics().frame_drop_count( + shard::metrics::frame_drop_variant::CONSENSUS, + shard::metrics::frame_drop_reason::UNSUPPORTED_OPERATION, + ); + shard.dispatch(message); + assert_eq!( + shard.metrics().frame_drop_count( + shard::metrics::frame_drop_variant::CONSENSUS, + shard::metrics::frame_drop_reason::UNSUPPORTED_OPERATION, + ), + before + 1, + "an operation from a newer release must be accounted under its own reason, not \ + folded into the generic unparsable drop" + ); + assert_eq!( + shard.metrics().frame_drop_count( + shard::metrics::frame_drop_variant::CONSENSUS, + shard::metrics::frame_drop_reason::UNPARSABLE, + ), + 0, + "version skew must not read as header corruption" + ); + } + /// Receive half of the purge gate in `on_repair_range_reply`: while a /// committed purge has not applied locally, a repair verdict must be /// deferred wholesale -- installing the peer's floor against pre-purge diff --git a/core/server_common/src/consensus_message.rs b/core/server_common/src/consensus_message.rs index cd4cd91e1..391790d0a 100644 --- a/core/server_common/src/consensus_message.rs +++ b/core/server_common/src/consensus_message.rs @@ -242,7 +242,7 @@ where let bytes = <B as MessageBacking<T>>::header_storage(&self.backing); let typed = bytemuck::checked::try_from_bytes::<T>(&bytes[..size_of::<T>()]) - .map_err(|_| ConsensusError::InvalidBitPattern)?; + .map_err(|_| classify_failed_cast::<T>(bytes))?; // Before `validate`: a header that did not survive the link intact cannot // have any of its fields believed, and `validate` reads them. typed.verify_frame()?; @@ -281,7 +281,7 @@ where let bytes = <B as MessageBacking<T>>::header_storage(&self.backing); let typed = bytemuck::checked::try_from_bytes::<T>(&bytes[..size_of::<T>()]) - .map_err(|_| ConsensusError::InvalidBitPattern)?; + .map_err(|_| classify_failed_cast::<T>(bytes))?; // Before `validate`: a header that did not survive the link intact cannot // have any of its fields believed, and `validate` reads them. typed.verify_frame()?; @@ -477,7 +477,7 @@ where } let header = bytemuck::checked::try_from_bytes::<H>(&bytes[..size_of::<H>()]) - .map_err(|_| ConsensusError::InvalidBitPattern)?; + .map_err(|_| classify_failed_cast::<H>(bytes))?; header.validate()?; // `size` is the whole-frame length and must at least span the header, or @@ -524,7 +524,7 @@ where } let header = bytemuck::checked::try_from_bytes::<H>(&first.as_slice()[..size_of::<H>()]) - .map_err(|_| ConsensusError::InvalidBitPattern)?; + .map_err(|_| classify_failed_cast::<H>(first.as_slice()))?; header.validate()?; // See `TryFrom<Owned>`: `size` must at least span the header so a @@ -774,6 +774,25 @@ impl MessageBag { } } +/// Why `H`'s header bytes failed bytemuck's checked cast. +/// +/// An operation discriminant this build does not define means the sender runs a +/// newer release; the frame is wire-valid and the node needs upgrading, which is +/// a different operator action from the corrupted-header case. bytemuck reports +/// both as one error, so the operation byte is probed here to separate them. +fn classify_failed_cast<H>(bytes: &[u8]) -> ConsensusError +where + H: ConsensusHeader, +{ + if let Some(offset) = H::OPERATION_OFFSET + && let Some(&code) = bytes.get(offset) + && !Operation::is_known_code(code) + { + return ConsensusError::UnsupportedOperation { operation: code }; + } + ConsensusError::InvalidBitPattern +} + impl<T> TryFrom<Message<T>> for MessageBag where T: ConsensusHeader, diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs index 99f70a224..6d2e725e6 100644 --- a/core/shard/src/metrics.rs +++ b/core/shard/src/metrics.rs @@ -120,6 +120,14 @@ pub mod frame_drop_variant { /// replicated traffic has nobody to answer, so this is the only record the op /// was destroyed. pub mod frame_drop_reason { + /// Operation discriminant unknown to this build: the sender is newer. + /// + /// Distinct from `UNPARSABLE` because upgrading this node is the fix, and + /// until it is, the frame's consensus group gap-stops here. + pub const UNSUPPORTED_OPERATION: &str = "unsupported_operation"; + /// A consensus frame failed typed decode for any other reason (corrupt + /// header, bad size, client-bound command on the inbound path). + pub const UNPARSABLE: &str = "unparsable"; pub const FULL: &str = "full"; pub const DISCONNECTED: &str = "disconnected"; pub const UNROUTABLE: &str = "unroutable"; @@ -134,7 +142,7 @@ pub mod frame_drop_reason { // site actually produces it, so the unreachable corners of the 7 x 7 cross // product never appear as permanent zero-valued series. const VARIANT_COUNT: usize = 7; -const REASON_COUNT: usize = 7; +const REASON_COUNT: usize = 9; const VARIANTS: [&str; VARIANT_COUNT] = [ frame_drop_variant::CONSENSUS, @@ -147,6 +155,8 @@ const VARIANTS: [&str; VARIANT_COUNT] = [ ]; const REASONS: [&str; REASON_COUNT] = [ + frame_drop_reason::UNSUPPORTED_OPERATION, + frame_drop_reason::UNPARSABLE, frame_drop_reason::FULL, frame_drop_reason::DISCONNECTED, frame_drop_reason::UNROUTABLE, diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs index 4340761a0..a54fce9c0 100644 --- a/core/shard/src/router.rs +++ b/core/shard/src/router.rs @@ -23,7 +23,7 @@ use crate::{IggyShard, LifecycleFrame, Receiver, RestorableMetadataStm, ShardFra use consensus::{MetadataHandle, PartitionsHandle}; use crossfire::TrySendError; use futures::FutureExt; -use iggy_binary_protocol::{GenericHeader, Operation, PrepareHeader}; +use iggy_binary_protocol::{ConsensusError, GenericHeader, Operation, PrepareHeader}; use journal::superblock::SuperblockStore; use journal::{Journal, JournalHandle}; use message_bus::{ConnectionInstaller, MessageBus, ReplicaHandshakeDoneFn}; @@ -62,17 +62,37 @@ where pub fn dispatch(&self, message: Message<GenericHeader>) { let bag = match MessageBag::try_from(message) { Ok(bag) => bag, + Err(ConsensusError::UnsupportedOperation { operation }) => { + // Terminal for this consensus group, not a per-frame hiccup: + // the op is never journaled, never acked, and every later + // prepare dies on the resulting gap while quorum hides the + // outage. Repair wraps the same typed header, so it cannot + // rescue this node either -- only upgrading it can. + // + // TODO(hubcio): the peer is still neither told nor fenced. + // A replica-plane version fence (release_min / release_max + // bounds, refusing the handshake outside them) is what makes a + // mixed-version cluster safe across a consensus-op addition; + // `PrepareHeader::release` already carries the sender's build. + self.metrics.record_frame_drop( + frame_drop_variant::CONSENSUS, + frame_drop_reason::UNSUPPORTED_OPERATION, + ); + tracing::error!( + shard = self.id, + operation = format_args!("{operation:#04x}"), + build_release = iggy_binary_protocol::IGGY_PROTOCOL_VERSION, + "consensus frame carries an operation this build does not know; the sender \ + runs a newer release. This node cannot journal or ack it, so its consensus \ + group stops making progress until this node is upgraded" + ); + return; + } Err(e) => { - // TODO(hubcio): this drop is the whole story for a consensus - // frame carrying an Operation this build does not know: no - // metric, no peer error, no eviction. An old node in a mixed - // cluster silently gap-stops the group here (never journals - // the op, never PrepareOks, every later prepare dies on the - // gap check) while quorum hides it, and repair wraps the same - // typed header so it cannot rescue. Rolling upgrades across - // consensus-op additions need a version fence (release_min / - // release_max bounds on the replica plane) before this arm is - // safe to hit. + self.metrics.record_frame_drop( + frame_drop_variant::CONSENSUS, + frame_drop_reason::UNPARSABLE, + ); tracing::warn!(shard = self.id, error = %e, "dropping unparsable consensus frame"); return; } @@ -804,25 +824,19 @@ mod tests { use server_common::{MESSAGE_ALIGN, Message, MessageBag}; use std::mem::offset_of; - /// RED SPEC, expected to FAIL: pins the mixed-cluster upgrade hole in - /// `dispatch`'s decode seam. - /// /// An `Operation` discriminant this build does not know, arriving on an /// otherwise wire-valid consensus frame (correct command, size, checksum: - /// exactly what a newer release sends after an op addition), decodes to - /// the same undifferentiated `ConsensusError::InvalidBitPattern` as random - /// memory corruption. `dispatch` answers both identically: a warn log and - /// a dropped frame. No metric, no peer error, no eviction, no version - /// fence. An old node in a mixed cluster therefore gap-stops its consensus - /// group silently (never journals the op, never sends a `PrepareOk`, every - /// later prepare dies on the gap check) while quorum hides the outage. + /// exactly what a newer release sends after an op addition), must decode to + /// its own error rather than the `InvalidBitPattern` that random memory + /// corruption produces. /// - /// Passes once the decode surfaces a dedicated unsupported-operation - /// signal the router can fence and account, instead of collapsing it into - /// the corruption error. + /// The two need different operator actions: version skew is fixed by + /// upgrading this node, and until it is, the frame's consensus group makes + /// no progress (the op is never journaled, never acked, and every later + /// prepare dies on the gap). `dispatch` splits its drop arms on this + /// distinction; the accounting half is pinned in the server crate by + /// `given_an_unknown_operation_when_dispatched_should_account_an_upgrade_fence_drop`. #[test] - // TODO(hubcio): fix this test - #[ignore = "unknown operation collapses into InvalidBitPattern; no upgrade fence"] fn given_an_unknown_operation_when_a_consensus_frame_decodes_should_surface_an_upgrade_fence_signal() { // Far past every defined Operation discriminant (the highest is 165). @@ -858,7 +872,12 @@ mod tests { }; assert!( - !matches!(error, ConsensusError::InvalidBitPattern), + matches!( + error, + ConsensusError::UnsupportedOperation { + operation: OPERATION_FROM_A_NEWER_RELEASE + } + ), "unknown operation {OPERATION_FROM_A_NEWER_RELEASE:#x} is silently dropped: the \ typed decode collapses a wire-valid frame from a newer release into the same \ InvalidBitPattern as corruption, and dispatch drops both with only a warn log, \
