You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/08/20 14:31:47 UTC
(iggy) branch fix_version_fencing created (now 47a015b59)
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a change to branch fix_version_fencing
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 47a015b59 fix(shard): tell an unknown consensus operation from a corrupt header
This branch includes the following new commits:
new 47a015b59 fix(shard): tell an unknown consensus operation from a corrupt header
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: fix(shard): tell an unknown consensus operation from a corrupt header
Posted by gk...@apache.org.
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 <g....@gmail.com>
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, \