This is an automated email from the ASF dual-hosted git repository.
spetz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new ff2e7481c fix(shard): refuse a frame no consensus plane claims (#4103)
ff2e7481c is described below
commit ff2e7481cd66f70cfca7dcbbcdf4d7282a8b5738
Author: Krishna Vishal <[email protected]>
AuthorDate: Fri Sep 11 20:45:43 2026 +0530
fix(shard): refuse a frame no consensus plane claims (#4103)
Shard dispatch silently ignores requests that neither consensus plane
accepts, leaving the client waiting until its read timeout.
`DeleteSegments` and `NonReplicated` fall into this gap. Released builds
avoid it because `classify` handles both before dispatch, as checked by
`classify_pins_probe_order`.
Operation routing is defined separately in `classify`, `route_typed`,
and the two plane predicates. These definitions already differ, so a
future operation or caller could reach the silent fallback.
This change adds `is_plane_routable` checks to the request, prepare, and
prepare-ack branches of `dispatch_message`:
* Requests receive a permanent `InvalidCommand`; retrying cannot fix an
unsupported operation.
* Prepares and acks are dropped, counted as
`frame_drops_total{variant=consensus,reason=unroutable}`, and logged
with frame details.
The checks stay inside those branches because other messages use
`Reserved` and have dedicated handlers.
An operation-coverage test checks every known code, rejects overlapping
plane ownership, and lists the three excluded operations with their
handling paths. The workload runner also checks `unroutable` and
`misrouted` counters each tick, reporting the shard and reason
immediately instead of waiting for the quiescence timeout.
---
core/binary_protocol/src/consensus/operation.rs | 40 ++++++++
core/server/src/dispatch/partition.rs | 118 +++++++++++++++++++++++-
core/shard/src/lib.rs | 105 ++++++++++++++++++++-
core/shard/src/metrics.rs | 43 +++++++++
core/simulator/src/workload/invariants.rs | 30 ++++++
5 files changed, 331 insertions(+), 5 deletions(-)
diff --git a/core/binary_protocol/src/consensus/operation.rs
b/core/binary_protocol/src/consensus/operation.rs
index 95de1d4ee..8ead254e5 100644
--- a/core/binary_protocol/src/consensus/operation.rs
+++ b/core/binary_protocol/src/consensus/operation.rs
@@ -203,6 +203,14 @@ impl Operation {
self.is_metadata() || matches!(self, Self::Register | Self::Logout)
}
+ /// Whether a consensus plane claims this operation. Anything else the
plane
+ /// chain drops in its `()` terminator, so entry paths must refuse it
first.
+ #[must_use]
+ #[inline]
+ pub const fn is_plane_routable(&self) -> bool {
+ self.is_metadata_plane() || self.is_partition()
+ }
+
/// Operations clients are allowed to send directly.
#[must_use]
#[inline]
@@ -368,4 +376,36 @@ mod tests {
assert!(Operation::StoreConsumerOffset.is_partition());
assert!(Operation::DeleteConsumerOffset.is_partition());
}
+
+ /// Every operation belongs to exactly one plane, or to the short list
+ /// answered before the chain. Walks `is_known_code` so a new variant fails
+ /// here rather than on the first frame carrying it.
+ #[test]
+ fn unroutable_operations_are_listed() {
+ // Answered earlier: `Reserved` by `validate_request_fields`,
+ // `NonReplicated` by the reads router, `DeleteSegments` by resolution
to
+ // `TruncatePartition` (`server::dispatch::classify`).
+ const UNROUTABLE: [Operation; 3] = [
+ Operation::Reserved,
+ Operation::NonReplicated,
+ Operation::DeleteSegments,
+ ];
+
+ for code in 0..=u8::MAX {
+ if !Operation::is_known_code(code) {
+ continue;
+ }
+ let operation: Operation = bytemuck::checked::cast(code);
+ assert!(
+ !(operation.is_metadata_plane() && operation.is_partition()),
+ "{operation:?} is claimed by both planes; the chain takes the
first"
+ );
+ assert_eq!(
+ operation.is_plane_routable(),
+ !UNROUTABLE.contains(&operation),
+ "{operation:?}: is_plane_routable={} disagrees with the
unroutable list",
+ operation.is_plane_routable(),
+ );
+ }
+ }
}
diff --git a/core/server/src/dispatch/partition.rs
b/core/server/src/dispatch/partition.rs
index de74599e0..dbed56cfa 100644
--- a/core/server/src/dispatch/partition.rs
+++ b/core/server/src/dispatch/partition.rs
@@ -1411,7 +1411,6 @@ mod tests {
};
#[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;
use iggy_binary_protocol::requests::messages::SendMessagesHeader;
@@ -1419,6 +1418,7 @@ mod tests {
use iggy_binary_protocol::requests::topics::{
CreateTopicRequest, CreateTopicWithAssignmentsRequest,
};
+ use iggy_binary_protocol::{PrepareOkHeader, ReplyHeader};
use iggy_binary_protocol::{WireName, WireOptions, WirePartitioning};
use iggy_common::Identifier;
use iggy_common::defaults::DEFAULT_ROOT_USER_ID;
@@ -1427,7 +1427,7 @@ mod tests {
use partitions::{IggyPartitions, PartitionPathLayout, PartitionsConfig};
use server_common::MessageBag;
use server_common::sharding::ShardId;
- use shard::metrics::ShardMetrics;
+ use shard::metrics::{ShardMetrics, frame_drop_reason, frame_drop_variant};
use shard::shards_table::PapayaShardsTable;
use shard::{
LifecycleFrame, PartitionConsensusConfig, ReconcileOp,
ReplicaTopology, ShardFrame,
@@ -2011,6 +2011,120 @@ mod tests {
);
}
+ /// A request no plane claims must be denied at the shard boundary, not
left
+ /// to the chain terminator and the client's read-timeout. `DeleteSegments`
+ /// is the live example: dispatch resolves it to `TruncatePartition`, so it
+ /// satisfies neither plane predicate. The deny is permanent -- a transient
+ /// code is what the SDK replays.
+ #[compio::test]
+ async fn unroutable_operation_must_reply_denied_not_silence() {
+ const TRANSPORT: u128 = 91;
+ const SESSION: u64 = 1;
+ const STATUS_OFFSET: usize = std::mem::offset_of!(ReplyHeader, status);
+
+ let bus = SpyBus::default();
+ let metadata = IggyMetadata::new(None, None, None, None,
TestMux::default(), None);
+ let partitions = IggyPartitions::new(
+ ShardId::new(0),
+ PartitionsConfig {
+ messages_required_to_save: 1,
+ size_of_messages_required_to_save:
iggy_common::IggyByteSize::from(1024_u64),
+ validate_checksum: true,
+ segment_size: iggy_common::IggyByteSize::from(1_048_576_u64),
+ preallocate_segments: false,
+ encryptor: None,
+ path_layout: PartitionPathLayout::default(),
+ },
+ );
+ let shard = Rc::new(TestShard::without_inbox(
+ ShardIdentity::new(0, "unroutable-operation-test".to_string()),
+ bus.clone(),
+ metadata,
+ partitions,
+ PapayaShardsTable::new(),
+ PartitionConsensusConfig::new(1, ReplicaTopology::new(0, 1),
bus.clone()),
+ ));
+
+ let request = request_message(Operation::DeleteSegments, TRANSPORT,
SESSION, 1, &[]);
+ shard.on_message(MessageBag::Request(request)).await;
+
+ let replies = bus.client_replies.borrow();
+ assert_eq!(
+ replies.len(),
+ 1,
+ "a request no plane claims must be answered, not absorbed in
silence"
+ );
+ let (client, frame) = &replies[0];
+ assert_eq!(*client, TRANSPORT, "reply must target the request's
client");
+ let status =
+ u32::from_le_bytes(frame[STATUS_OFFSET..STATUS_OFFSET +
4].try_into().unwrap());
+ assert_eq!(
+ status,
+ IggyError::InvalidCommand.as_code(),
+ "the deny must be permanent; a transient status has the SDK replay
it"
+ );
+ }
+
+ /// The replicated halves of the same hole. A prepare or an ack whose
+ /// operation no plane claims reaches the same terminator, and there is no
+ /// client to answer, so the counter and the log are the only record.
Dropping
+ /// it is the only option: nothing journals or acks an operation no plane
+ /// owns.
+ #[compio::test]
+ async fn unroutable_replicated_frames_must_be_dropped_and_counted() {
+ const TRANSPORT: u128 = 91;
+
+ let bus = SpyBus::default();
+ let shard = Rc::new(test_shard(&bus, 0, 1, 1));
+ let unroutable_drops = || {
+ shard
+ .metrics()
+ .frame_drop_count(frame_drop_variant::CONSENSUS,
frame_drop_reason::UNROUTABLE)
+ };
+
+ let prepare = prepare_message(Operation::DeleteSegments, TRANSPORT, 1,
&[]);
+ shard.on_message(MessageBag::Prepare(prepare)).await;
+ assert_eq!(
+ unroutable_drops(),
+ 1,
+ "a prepare no plane claims must be counted, not absorbed by the
chain"
+ );
+
+ shard
+ .on_message(MessageBag::PrepareOk(prepare_ok_message(
+ Operation::DeleteSegments,
+ 1,
+ )))
+ .await;
+ assert_eq!(
+ unroutable_drops(),
+ 2,
+ "an ack no plane claims must be counted, not absorbed by the chain"
+ );
+
+ assert!(
+ bus.client_replies.borrow().is_empty(),
+ "replicated frames answer nobody on this node"
+ );
+ }
+
+ /// Bare `PrepareOk` for the routing guard: only `operation`, `group` and
`op`
+ /// are read before the plane chain, so the rest stays zeroed.
+ fn prepare_ok_message(operation: Operation, op: u64) ->
Message<PrepareOkHeader> {
+ let header_size = size_of::<PrepareOkHeader>();
+ let mut msg = Message::<PrepareOkHeader>::new(header_size);
+ let header = bytemuck::checked::try_from_bytes_mut::<PrepareOkHeader>(
+ &mut msg.as_mut_slice()[..header_size],
+ )
+ .expect("zeroed bytes form a valid PrepareOkHeader");
+ header.command = Command::PrepareOk;
+ header.size = u32::try_from(header_size).expect("ack size fits u32");
+ header.operation = operation;
+ header.op = op;
+ header.group = server_common::sharding::METADATA_GROUP;
+ msg
+ }
+
/// A send parked for a namespace that is torn down before materialising
/// (create -> delete before the reconciler's `InsertOwned`) is discarded
/// on `ConfirmRemove`. The discard must stage the same retriable
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index ede938135..2a0be3f47 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -3053,7 +3053,7 @@ where
/// [`Self::on_message`] carrying the park provenance of a frame the pump
is
/// re-delivering, so a second park keeps the stamp and age the first one
/// derived instead of deriving them again against newer committed state.
- #[allow(clippy::future_not_send)]
+ #[allow(clippy::future_not_send, clippy::too_many_lines)]
async fn dispatch_message(&self, message: MessageBag, provenance:
Option<ParkProvenance>)
where
B: MessageBus + 'static,
@@ -3070,6 +3070,12 @@ where
let header = request.header();
(header.operation, header.group)
};
+ // Ahead of the park and the incarnation fence, which both read
+ // the frame as a partition request.
+ if !routing.0.is_plane_routable() {
+ self.deny_unroutable_request(request.header()).await;
+ return;
+ }
match self
.park_if_unmaterialised(request, routing.0, routing.1,
provenance, &mut None)
{
@@ -3098,8 +3104,17 @@ where
MessageBag::Prepare(prepare) => {
let routing = {
let header = prepare.header();
- (header.operation, header.group)
+ (header.operation, header.group, header.op)
};
+ if !routing.0.is_plane_routable() {
+ self.drop_unroutable_replicated(
+ Command::Prepare,
+ routing.0,
+ routing.1,
+ routing.2,
+ );
+ return;
+ }
// A tombstoned prepare still flows to the plane: replicated
// traffic has no client awaiting a reply on this node, and
// the plane's own tombstone guard drops it.
@@ -3140,7 +3155,22 @@ where
ParkOutcome::Overflow(_) | ParkOutcome::Parked => {}
}
}
- MessageBag::PrepareOk(prepare_ok) => self.on_ack(prepare_ok).await,
+ MessageBag::PrepareOk(prepare_ok) => {
+ let routing = {
+ let header = prepare_ok.header();
+ (header.operation, header.group, header.op)
+ };
+ if !routing.0.is_plane_routable() {
+ self.drop_unroutable_replicated(
+ Command::PrepareOk,
+ routing.0,
+ routing.1,
+ routing.2,
+ );
+ return;
+ }
+ self.on_ack(prepare_ok).await;
+ }
MessageBag::StartViewChange(msg) =>
self.on_start_view_change(msg).await,
MessageBag::DoViewChange(msg) => self.on_do_view_change(msg).await,
MessageBag::StartView(msg) => self.on_start_view(msg).await,
@@ -3796,6 +3826,75 @@ where
self.metrics.record_partition_request_denied_transient();
}
+ /// Deny a request no consensus plane claims, and count the frame it drops.
+ /// `InvalidCommand` because no retry makes an operation routable. Counted
as
+ /// a drop even though the client is answered: nothing was routed or
+ /// journaled, and `unroutable` is the counter a simulator run asserts on.
+ #[allow(clippy::future_not_send)]
+ async fn deny_unroutable_request(&self, request_header:
&RoutedRequestHeader) {
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::CONSENSUS,
+ crate::metrics::frame_drop_reason::UNROUTABLE,
+ );
+ tracing::error!(
+ shard = self.id,
+ client = request_header.client,
+ operation = ?request_header.operation,
+ namespace_raw = request_header.group,
+ "request operation is claimed by no consensus plane; denying it"
+ );
+ let reply = build_deny_reply_from_request_header(
+ request_header,
+ IggyError::InvalidCommand.as_code(),
+ );
+ if let Err(error) = self
+ .bus
+ .send_to_client(request_header.client,
reply.into_generic().into_frozen())
+ .await
+ {
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::CONSENSUS,
+ crate::metrics::frame_drop_reason::DELIVERY_FAILED,
+ );
+ tracing::warn!(
+ shard = self.id,
+ client = request_header.client,
+ operation = ?request_header.operation,
+ error = %error,
+ "failed to send deny for unroutable request"
+ );
+ }
+ }
+
+ /// Drop a replicated frame no consensus plane claims, and count it.
+ ///
+ /// No reply, unlike [`Self::deny_unroutable_request`]: a prepare or an ack
+ /// has no client waiting on this node. Terminal for the frame's group
here,
+ /// as the unknown-discriminant drop in [`Self::dispatch`] is: nothing
+ /// journals or acks an operation no plane owns, so every later op in that
+ /// group waits behind the gap while quorum hides it. Nothing fences the
+ /// sending peer, so the counter and this log are the whole signal.
+ fn drop_unroutable_replicated(
+ &self,
+ command: Command,
+ operation: Operation,
+ namespace_raw: u64,
+ op: u64,
+ ) {
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::CONSENSUS,
+ crate::metrics::frame_drop_reason::UNROUTABLE,
+ );
+ tracing::error!(
+ shard = self.id,
+ command = ?command,
+ operation = ?operation,
+ namespace_raw,
+ op,
+ "replicated frame operation is claimed by no consensus plane;
dropping it"
+ );
+ }
+
/// [`Self::deny_partition_request_transient`] for synchronous callers:
/// hand the deny to this shard's own pump as a
/// [`LifecycleFrame::ForwardClientSend`], whose handler performs the bus
diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs
index 01569de1b..2ac87c6be 100644
--- a/core/shard/src/metrics.rs
+++ b/core/shard/src/metrics.rs
@@ -562,6 +562,25 @@ impl ShardMetrics {
.sum()
}
+ /// One `reason` summed over every variant, for the assertions that are
about
+ /// the reason alone: `unroutable` and `misrouted` must stay at zero
however
+ /// the drop was classified, while `full` and `disconnected` are what a
+ /// crashed inbox produces and cannot be asserted on. Listing variants at
the
+ /// call site would let a new drop site under an unlisted one escape.
+ /// Unknown reasons read as zero, as in [`Self::frame_drop_count`].
+ #[cfg(any(test, feature = "simulator"))]
+ #[must_use]
+ pub fn frame_drop_count_for_reason(&self, reason: &'static str) -> u64 {
+ let Some(reason_idx) = reason_index(reason) else {
+ return 0;
+ };
+ self.cached_counters
+ .iter()
+ .filter_map(|variant| variant[reason_idx].get())
+ .map(prometheus_client::metrics::counter::Counter::get)
+ .sum()
+ }
+
/// Snapshot of `partitions_materialised_total`. Test-only accessor;
/// production scrape goes through the prometheus registry.
#[cfg(test)]
@@ -844,6 +863,30 @@ mod tests {
);
}
+ #[test]
+ fn frame_drop_count_for_reason_sums_one_reason_across_variants() {
+ let metrics = ShardMetrics::for_shard();
+ metrics.record_frame_drop(frame_drop_variant::CONSENSUS,
frame_drop_reason::UNROUTABLE);
+ metrics.record_frame_drop(frame_drop_variant::PARTITION,
frame_drop_reason::UNROUTABLE);
+ metrics.record_frame_drop(frame_drop_variant::CONSENSUS,
frame_drop_reason::FULL);
+
+ assert_eq!(
+ metrics.frame_drop_count_for_reason(frame_drop_reason::UNROUTABLE),
+ 2,
+ "one reason sums across every variant that recorded it",
+ );
+ assert_eq!(
+ metrics.frame_drop_count_for_reason(frame_drop_reason::MISROUTED),
+ 0,
+ "an unproduced reason reads as zero, not as the total",
+ );
+ assert_eq!(
+ metrics.frame_drops_value(),
+ 3,
+ "the per-reason view must not disturb the total",
+ );
+ }
+
#[test]
fn persistence_scrape_distinguishes_retained_budget_from_wal_file_bytes() {
let metrics = ShardMetrics::for_shard();
diff --git a/core/simulator/src/workload/invariants.rs
b/core/simulator/src/workload/invariants.rs
index d2011bcb5..a51103e97 100644
--- a/core/simulator/src/workload/invariants.rs
+++ b/core/simulator/src/workload/invariants.rs
@@ -28,6 +28,7 @@ use crate::workload::{CLIENT_REQUEST_QUEUE_MAX, Workload};
use crate::{CommitHoldKind, CommitPrefixHole, Simulator};
use consensus::Consensus;
use server_common::sharding::IggyNamespace;
+use shard::metrics::frame_drop_reason;
use std::collections::HashMap;
/// Ticks a `Normal` metadata primary may sit behind its own recovery barrier
@@ -38,6 +39,12 @@ use std::collections::HashMap;
/// of slack: only a barrier nothing will ever lower trips it.
const RECOVERY_BARRIER_WEDGE_TICKS: u32 = 2_000;
+/// Frame-drop reasons that name a bug rather than a fault. Crashes produce
+/// `full` / `disconnected` / `delivery_failed` routinely; nothing the injector
+/// does can misplace a frame.
+const BUG_ONLY_FRAME_DROP_REASONS: [&str; 2] =
+ [frame_drop_reason::UNROUTABLE, frame_drop_reason::MISROUTED];
+
/// Ticks a replica may hold its commit walk below a MISSING op.
///
/// A promotion holds here legitimately while the journal walk clears its apply
@@ -85,6 +92,7 @@ impl Invariants {
///
/// Globally:
/// - total in-flight requests stay within the per-client queue ceiling,
+ /// - no shard has shed a frame for an `unroutable` or `misrouted` reason,
/// - live replicas agree on every committed metadata op they share, and
the
/// committed chain stays hash-linked (see [`StateChecker`]).
///
@@ -145,6 +153,7 @@ impl Invariants {
(client_count={}, queue_max={CLIENT_REQUEST_QUEUE_MAX})
(seed={seed:#x})",
workload.options.client_count,
);
+ assert_no_bug_frame_drops(sim, seed);
self.state_checker.check(sim, seed);
}
@@ -273,6 +282,27 @@ impl Invariants {
}
}
+/// Panic if any shard has shed a frame for a reason no fault can produce.
+///
+/// Every tick rather than at quiesce, crashed replicas included: a restart
+/// rebuilds the shard with zero counters, so a drop is only visible on the
ticks
+/// between it and the next crash. Miss it and the run instead fails tens of
+/// thousands of ticks later as a drain that never completed.
+fn assert_no_bug_frame_drops(sim: &Simulator, seed: u64) {
+ for (replica_idx, replica) in sim.replicas.iter().enumerate() {
+ for (shard_idx, shard) in replica.shards.iter().enumerate() {
+ for reason in BUG_ONLY_FRAME_DROP_REASONS {
+ let dropped =
shard.metrics().frame_drop_count_for_reason(reason);
+ assert_eq!(
+ dropped, 0,
+ "replica {replica_idx} shard {shard_idx} shed {dropped}
frame(s) with \
+ reason={reason}; no injected fault produces that
(seed={seed:#x})"
+ );
+ }
+ }
+ }
+}
+
/// Panic if `cur < prev`. Pure so the catch logic is unit-testable without a
/// full simulator; shared by the `commit_offset` and `view` checks.
fn assert_no_regression(
