You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/07/28 10:52:10 UTC
(iggy) branch feat/command-extensibility created (now b49eaf767)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a change to branch feat/command-extensibility
in repository https://gitbox.apache.org/repos/asf/iggy.git
at b49eaf767 fix(sdk): forward unknown command codes instead of rejecting them
This branch includes the following new commits:
new b49eaf767 fix(sdk): forward unknown command codes instead of rejecting them
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(sdk): forward unknown command codes instead of rejecting them
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/command-extensibility
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit b49eaf7675d6df7c757be7267b9c88ccd0995328
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Tue Jul 28 12:25:25 2026 +0200
fix(sdk): forward unknown command codes instead of rejecting them
An application built on this SDK cannot reach a command its SDK build
has never heard of. operation_for_code ended None => InvalidCommand,
so a code absent from COMMAND_TABLE failed at encode time, before any
byte left the process, and a server that does implement it never got
the chance to answer. COMMAND_TABLE is a protocol registry, not a
per-server capability list, so the server is the authority on whether
a code exists: an unknown code now ships as non-replicated, carrying
the code in RequestHeader.reserved, and the server answers with a
proper error if it does not know it. A code the registry knows as
replicated is still refused, since sending one non-replicated would
apply it on a single node and silently diverge the replicas.
The client-side namespace derivation goes with it. The server
re-derives the namespace from the request body, and resolves named
identifiers and Balanced or MessagesKey partitioning that the SDK
refused outright, so the client copy was redundant and weaker. Not a
pure deletion: out-of-range stream, topic and partition ids are now
rejected by the server rather than before the request leaves the
process, and a partition op using Balanced or MessagesKey no longer
fails fast client-side.
Letting requests through exposed three ways an operation could fail
quietly once it arrived.
An Operation no state machine claimed reached the mux terminus, which
returned ApplyReply::default(). That is code 0, the defined success
value, so the client was told its write succeeded while no replica
had applied anything, and the commit paths' panic for a committed op
that cannot be applied never fired, because the walk returned Ok. The
terminus now returns Err: on_ack and commit_journal panic, and WAL
replay surfaces RecoveryError::StateMachine. A committed op with no
handler is state-machine divergence, not a client error. The
permissive base case had one consumer, the recovery test harness,
which gets its own terminus so production stays strict.
A frame whose operation matched no plane was caught by a debug_assert
that compiles out in release, letting the frame fall through into
route_consensus_control, where it was injected onto an arbitrary
shard as a bogus VSR control message. It is now counted, logged and
dropped.
Reaching that fall-through was easier than it looks, because
is_metadata mixed a range test with a hand-written 21-variant match:
a discriminant in the metadata band's own free tail classified as
neither plane. It is now a range over the same bounds, with
DeleteSegments carved out, since the dispatch layer rewrites it into
a TruncatePartition before consensus and it never reaches a plane
under its own discriminant. Classification is unchanged for every
assigned discriminant.
---
core/binary_protocol/src/consensus/operation.rs | 82 ++++++---
core/integration/tests/sdk/raw.rs | 13 +-
core/metadata/src/impls/recovery.rs | 3 +-
core/metadata/src/stm/authz.rs | 40 +++-
core/metadata/src/stm/mux.rs | 37 +++-
core/sdk/src/clients/client.rs | 8 +-
core/sdk/src/vsr.rs | 233 +++++++-----------------
core/shard/src/metrics.rs | 6 +-
core/shard/src/router.rs | 174 ++++++++++++------
9 files changed, 332 insertions(+), 264 deletions(-)
diff --git a/core/binary_protocol/src/consensus/operation.rs b/core/binary_protocol/src/consensus/operation.rs
index 11d15a106..3c14b9209 100644
--- a/core/binary_protocol/src/consensus/operation.rs
+++ b/core/binary_protocol/src/consensus/operation.rs
@@ -77,7 +77,7 @@ pub enum Operation {
// Client op handled specially: the dispatch layer resolves the requested
// segment count to an offset on the owning shard and replicates a
// `TruncatePartition` through metadata, so `DeleteSegments` itself is
- // neither a metadata nor a partition consensus op (see `is_partition`).
+ // neither a metadata nor a partition consensus op (see `is_metadata`).
DeleteSegments = 138,
CreateConsumerGroup = 139,
DeleteConsumerGroup = 140,
@@ -112,37 +112,17 @@ impl Operation {
}
/// Metadata / control-plane operations handled by shard 0.
+ ///
+ /// `DeleteSegments` is the one discriminant in the range that is excluded:
+ /// the dispatch layer rewrites it into a `TruncatePartition` before
+ /// consensus, so it never reaches a plane under its own discriminant.
#[must_use]
#[inline]
pub const fn is_metadata(&self) -> bool {
- if self.is_internal() {
- return true;
- }
-
- matches!(
- self,
- Self::CreateStream
- | Self::UpdateStream
- | Self::DeleteStream
- | Self::PurgeStream
- | Self::CreateTopic
- | Self::UpdateTopic
- | Self::DeleteTopic
- | Self::PurgeTopic
- | Self::CreatePartitions
- | Self::DeletePartitions
- | Self::CreateConsumerGroup
- | Self::DeleteConsumerGroup
- | Self::CreateUser
- | Self::UpdateUser
- | Self::DeleteUser
- | Self::ChangePassword
- | Self::UpdatePermissions
- | Self::CreatePersonalAccessToken
- | Self::DeletePersonalAccessToken
- | Self::JoinConsumerGroup
- | Self::LeaveConsumerGroup
- )
+ self.is_internal()
+ || ((*self as u8) >= Self::METADATA_START
+ && (*self as u8) < Self::PARTITION_START
+ && !matches!(self, Self::DeleteSegments))
}
/// VSR protocol-level operations that are not mapped directly to a
@@ -257,6 +237,50 @@ impl Operation {
mod tests {
use super::Operation;
+ /// Every discriminant the enum assigns, taken from the `CheckedBitPattern`
+ /// whitelist so it cannot drift from the variant list.
+ fn assigned_operations() -> Vec<Operation> {
+ (0..=u8::MAX)
+ .filter_map(|code| bytemuck::checked::try_cast::<u8, Operation>(code).ok())
+ .collect()
+ }
+
+ /// A metadata variant added in the band's unallocated tail must classify as
+ /// metadata. It cannot be constructed here (`CheckedBitPattern` rejects an
+ /// unassigned discriminant), so the tail is pinned against the bounds
+ /// `is_metadata` tests and against the one carved-out discriminant.
+ #[test]
+ fn metadata_band_free_tail_falls_inside_metadata_bounds() {
+ let metadata_bounds = Operation::METADATA_START..Operation::PARTITION_START;
+ let free_tail = (Operation::LeaveConsumerGroup as u8 + 1)..Operation::PARTITION_START;
+ assert_eq!(free_tail, 150..160);
+
+ for code in free_tail {
+ assert!(
+ metadata_bounds.contains(&code),
+ "discriminant {code} is outside the metadata bounds"
+ );
+ assert_ne!(
+ code,
+ Operation::DeleteSegments as u8,
+ "discriminant {code} collides with the carve-out"
+ );
+ }
+ }
+
+ #[test]
+ fn every_assigned_discriminant_belongs_to_exactly_one_plane() {
+ for op in assigned_operations() {
+ let planes = u8::from(op.is_vsr_reserved())
+ + u8::from(op.is_metadata())
+ + u8::from(op.is_partition());
+ // `DeleteSegments` is the single deliberate hole: dispatch rewrites
+ // it into `TruncatePartition` before it can reach a plane.
+ let expected = u8::from(op != Operation::DeleteSegments);
+ assert_eq!(planes, expected, "{op:?} is claimed by {planes} planes");
+ }
+ }
+
#[test]
fn command_code_roundtrip() {
let ops = [
diff --git a/core/integration/tests/sdk/raw.rs b/core/integration/tests/sdk/raw.rs
index cffeec954..0bf004947 100644
--- a/core/integration/tests/sdk/raw.rs
+++ b/core/integration/tests/sdk/raw.rs
@@ -87,14 +87,23 @@ async fn assert_raw_round_trip(client: &IggyClient) {
.expect_err("session-control codes must be rejected by the raw path");
assert_eq!(error, IggyError::InvalidCommand);
- // VSR encoder is closed-world: unknown code rejected at encode time.
+ // An SDK build cannot know which codes a server implements, so an
+ // unknown one ships as non-replicated and the server decides. The
+ // follow-up ping is the point of the case: the server must answer
+ // with a deny frame, not drop the frame and leave the connection
+ // wedged until the read timeout.
#[cfg(feature = "vsr")]
{
let error = client
.send_binary_request(60_000, Bytes::new())
.await
- .expect_err("unknown code must be rejected under VSR");
+ .expect_err("unknown code must be refused by the server");
assert_eq!(error, IggyError::InvalidCommand);
+
+ client
+ .send_binary_request(PING_CODE, PingRequest.to_bytes())
+ .await
+ .expect("connection must survive an unknown code");
}
let error = client
diff --git a/core/metadata/src/impls/recovery.rs b/core/metadata/src/impls/recovery.rs
index 00ea8b800..9d6630a2a 100644
--- a/core/metadata/src/impls/recovery.rs
+++ b/core/metadata/src/impls/recovery.rs
@@ -306,8 +306,9 @@ mod tests {
use tempfile::tempdir;
use crate::MuxStateMachine;
+ use crate::stm::mux::NoopState;
- type TestStm = MuxStateMachine<()>;
+ type TestStm = MuxStateMachine<NoopState>;
const HEADER_SIZE: usize = size_of::<PrepareHeader>();
diff --git a/core/metadata/src/stm/authz.rs b/core/metadata/src/stm/authz.rs
index 00ea17416..82048245c 100644
--- a/core/metadata/src/stm/authz.rs
+++ b/core/metadata/src/stm/authz.rs
@@ -37,6 +37,8 @@ use crate::impls::metadata::StreamsFrontend;
use crate::permissioner::Permissioner;
use crate::stm::StateMachine;
use crate::stm::consumer_group::{JoinConsumerGroupRequest, LeaveConsumerGroupRequest};
+#[cfg(test)]
+use crate::stm::mux::NoopState;
use crate::stm::result::ApplyReply;
use crate::stm::stream::{Streams, TruncatePartitionRequest};
use crate::stm::user::Users;
@@ -389,11 +391,12 @@ fn topic_scoped(
check(users, |perm| rule(perm, stream_id, topic_id))
}
-/// The recovery unit tests exercise WAL-replay mechanics with an empty state
-/// list (`MuxStateMachine<()>`), which has no permission state and so needs no
-/// gate. Production always uses the `variadic!(Users, Streams)` machine above.
+/// The recovery unit tests exercise WAL-replay mechanics with a state list that
+/// claims nothing (`MuxStateMachine<NoopState>`), which has no permission state
+/// and so needs no gate. Production always uses the `variadic!(Users, Streams)`
+/// machine above.
#[cfg(test)]
-impl GatedApply for MuxStateMachine<()> {
+impl GatedApply for MuxStateMachine<NoopState> {
fn gated_update(&self, prepare: Message<PrepareHeader>) -> Result<ApplyReply, IggyError> {
self.update(prepare)
}
@@ -682,6 +685,35 @@ mod tests {
);
}
+ // A committed op that walks the whole mux without any state claiming it
+ // must surface an error. `Ok` there would cache and ship an affirmative
+ // success reply for an op that mutated nothing on any replica.
+ // `CreateTopic` is the pre-projection wire form: the request path rewrites
+ // it to `CreateTopicWithAssignments`, so only `CreateTopicWithAssignments`
+ // has an apply handler.
+ #[test]
+ fn given_op_no_state_claims_when_applied_should_fail_not_reply_success() {
+ let mux = mux();
+ assert_eq!(
+ mux.gated_update(make_prepare(
+ 1,
+ Operation::CreateStream,
+ ROOT,
+ &create_stream_body("s1"),
+ ))
+ .unwrap()
+ .code,
+ 0,
+ "a claimed op still applies"
+ );
+
+ assert!(
+ mux.gated_update(make_prepare(2, Operation::CreateTopic, ROOT, &[]))
+ .is_err(),
+ "an unclaimed op must fail, never commit as a success reply"
+ );
+ }
+
#[test]
fn given_internal_revocation_op_when_authorized_should_skip_the_gate() {
let mux = mux();
diff --git a/core/metadata/src/stm/mux.rs b/core/metadata/src/stm/mux.rs
index 864d40e2f..70453c3be 100644
--- a/core/metadata/src/stm/mux.rs
+++ b/core/metadata/src/stm/mux.rs
@@ -132,17 +132,52 @@ where
}
}
-// Base case of the recursive resolution.
+// Base case of the recursive resolution: every state passed the op through, so
+// no state applies it. Committed-but-unappliable is a state-machine divergence,
+// not a client error, so this fails and lets the commit paths trip their
+// existing panic. Succeeding here would instead ship an affirmative success
+// reply (`ApplyReply::default()` carries `code == 0`) for an op that mutated
+// nothing on any replica.
impl StateMachine for () {
type Input = Message<PrepareHeader>;
type Output = crate::stm::result::ApplyReply;
type Error = iggy_common::IggyError;
+ fn update(&self, input: Self::Input) -> Result<Self::Output, Self::Error> {
+ tracing::error!(
+ operation = ?input.header().operation,
+ "committed operation is claimed by no state in the metadata mux"
+ );
+ Err(iggy_common::IggyError::InvalidCommand)
+ }
+}
+
+/// Permissive stand-in for the terminus above: claims nothing, succeeds anyway.
+/// Tests that drive the machinery around the state machine (mux plumbing, WAL
+/// replay) run on a state list that applies no op, which the strict terminus
+/// would reject.
+#[cfg(test)]
+#[derive(Debug, Default)]
+pub(crate) struct NoopState;
+
+#[cfg(test)]
+impl StateMachine for NoopState {
+ type Input = Message<PrepareHeader>;
+ type Output = crate::stm::result::ApplyReply;
+ type Error = iggy_common::IggyError;
+
fn update(&self, _input: Self::Input) -> Result<Self::Output, Self::Error> {
Ok(crate::stm::result::ApplyReply::default())
}
}
+#[cfg(test)]
+impl<S> RestoreSnapshot<S> for NoopState {
+ fn restore_snapshot(_snapshot: &S) -> Result<Self, SnapshotError> {
+ Ok(Self)
+ }
+}
+
// Recursive case: process head and recurse on tail
// No Clone bound needed - ownership passes through via `Either`
impl<O, E, S, Rest> StateMachine for variadic!(S, ...Rest)
diff --git a/core/sdk/src/clients/client.rs b/core/sdk/src/clients/client.rs
index 98beb9830..7cf03830d 100644
--- a/core/sdk/src/clients/client.rs
+++ b/core/sdk/src/clients/client.rs
@@ -204,9 +204,11 @@ impl IggyClient {
/// Login and logout codes are rejected with `InvalidCommand`. Use the
/// `login_user` / `logout_user` methods so SDK session state stays correct.
///
- /// Custom codes only work on the classic protocol. Under `vsr` the encoder
- /// is closed-world: an unknown code yields `InvalidCommand`, a replicated
- /// code with no mapping yields `UnknownReplicatedCommand`.
+ /// Custom codes are forwarded to the server, which is the authority on
+ /// whether it implements them. Under `vsr` they ship as non-replicated; a
+ /// code the registry knows as replicated but cannot map yields
+ /// `UnknownReplicatedCommand`, since replicating it non-replicated would
+ /// diverge the cluster.
pub async fn send_binary_request(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> {
if SESSION_CONTROL_CODES.contains(&code) {
return Err(IggyError::InvalidCommand);
diff --git a/core/sdk/src/vsr.rs b/core/sdk/src/vsr.rs
index 90a10a552..6eaf27900 100644
--- a/core/sdk/src/vsr.rs
+++ b/core/sdk/src/vsr.rs
@@ -17,27 +17,14 @@
use crate::session::ConsensusSession;
use bytes::{BufMut, Bytes, BytesMut};
-use iggy_binary_protocol::codec::WireDecode;
use iggy_binary_protocol::codes::{
- DELETE_CONSUMER_OFFSET_2_CODE, DELETE_CONSUMER_OFFSET_CODE, DELETE_SEGMENTS_CODE,
- LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE, LOGOUT_USER_CODE, SEND_MESSAGES_CODE,
- STORE_CONSUMER_OFFSET_2_CODE, STORE_CONSUMER_OFFSET_CODE,
+ LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE, LOGOUT_USER_CODE,
};
use iggy_binary_protocol::consensus::{
Command2, EvictionHeader, EvictionReason, GenericHeader, HEADER_SIZE, Operation, ReplyHeader,
RequestHeader, read_size_field, result_code, result_section_len,
};
-use iggy_binary_protocol::namespace::{
- MAX_PARTITIONS, MAX_STREAMS, MAX_TOPICS, METADATA_CONSENSUS_NAMESPACE, PARTITION_MASK,
- PARTITION_SHIFT, STREAM_MASK, STREAM_SHIFT, TOPIC_MASK, TOPIC_SHIFT,
-};
-use iggy_binary_protocol::requests::consumer_offsets::{
- DeleteConsumerOffset2Request, DeleteConsumerOffsetRequest, StoreConsumerOffset2Request,
- StoreConsumerOffsetRequest,
-};
-use iggy_binary_protocol::requests::messages::SendMessagesHeader;
-use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest;
-use iggy_binary_protocol::{WireIdentifier, WirePartitioning};
+use iggy_binary_protocol::namespace::METADATA_CONSENSUS_NAMESPACE;
use iggy_common::{IggyError, eviction_reason_to_error};
const NON_REPLICATED_CODE_RANGE: std::ops::Range<usize> = 0..4;
@@ -122,7 +109,18 @@ pub(crate) fn encode_request_header(
}
}
};
- let namespace = namespace_for_request(code, payload, operation)?;
+ // Control-plane requests target the metadata replica (shard 0). The
+ // router's `route_typed` only short-circuits to shard 0 when the
+ // namespace value equals `METADATA_CONSENSUS_NAMESPACE`; sending plain
+ // `0` falls into `route_consensus_control` which hashes the namespace
+ // and lands a Register on a peer shard whose `submit_register_in_process`
+ // panics ("consensus only exists on shard 0"). Everything else ships 0:
+ // the server re-derives the namespace from the request body, resolving
+ // named identifiers a client cannot.
+ let namespace = match operation {
+ Operation::Register | Operation::Logout => METADATA_CONSENSUS_NAMESPACE,
+ _ => 0,
+ };
let total_size = HEADER_SIZE
.checked_add(payload.len())
.ok_or(IggyError::InvalidConfiguration)?;
@@ -162,10 +160,17 @@ fn operation_for_code(code: u32) -> Result<Operation, IggyError> {
return Ok(operation);
}
+ // `COMMAND_TABLE` is a protocol registry, not a per-server capability list,
+ // so an SDK build cannot know which codes a given server implements. The
+ // server is the authority: forward an unknown code as non-replicated (the
+ // code rides `RequestHeader.reserved`, which that path already stamps) and
+ // let the server answer with a proper error if it does not know it. A
+ // known-but-replicated code still fails here -- sending it non-replicated
+ // would apply it on one node only and silently diverge the replicas.
match iggy_binary_protocol::dispatch::lookup_command(code) {
Some(meta) if !meta.is_replicated() => Ok(Operation::NonReplicated),
Some(_) => Err(IggyError::UnknownReplicatedCommand(code)),
- None => Err(IggyError::InvalidCommand),
+ None => Ok(Operation::NonReplicated),
}
}
@@ -345,123 +350,6 @@ fn read_window_field(header_bytes: &[u8; HEADER_SIZE], offset: usize) -> u32 {
u32::from_le_bytes(value)
}
-fn namespace_for_request(
- code: u32,
- payload: &Bytes,
- operation: Operation,
-) -> Result<u64, IggyError> {
- // Control-plane requests target the metadata replica (shard 0). The
- // router's `route_typed` only short-circuits to shard 0 when the
- // namespace value equals `METADATA_CONSENSUS_NAMESPACE`; sending plain
- // `0` falls into `route_consensus_control` which hashes the namespace
- // and lands a Register on a peer shard whose `submit_register_in_process`
- // panics ("consensus only exists on shard 0").
- if operation == Operation::Register || operation == Operation::Logout {
- return Ok(METADATA_CONSENSUS_NAMESPACE);
- }
- if operation == Operation::NonReplicated || operation.is_metadata() {
- return Ok(0);
- }
-
- let namespace = match code {
- SEND_MESSAGES_CODE => {
- if payload.len() < 4 {
- return Err(IggyError::InvalidCommand);
- }
- let metadata_length = u32::from_le_bytes(
- payload[..4]
- .try_into()
- .map_err(|_| IggyError::InvalidNumberEncoding)?,
- ) as usize;
- if payload.len() < 4 + metadata_length {
- return Err(IggyError::InvalidCommand);
- }
- let header = SendMessagesHeader::decode_from(&payload[4..4 + metadata_length])
- .map_err(|_| IggyError::InvalidCommand)?;
- namespace_from_partitioning(&header.stream_id, &header.topic_id, &header.partitioning)?
- }
- STORE_CONSUMER_OFFSET_CODE => {
- let request = StoreConsumerOffsetRequest::decode_from(payload)
- .map_err(|_| IggyError::InvalidCommand)?;
- namespace_from_partition(&request.stream_id, &request.topic_id, request.partition_id)?
- }
- DELETE_CONSUMER_OFFSET_CODE => {
- let request = DeleteConsumerOffsetRequest::decode_from(payload)
- .map_err(|_| IggyError::InvalidCommand)?;
- namespace_from_partition(&request.stream_id, &request.topic_id, request.partition_id)?
- }
- STORE_CONSUMER_OFFSET_2_CODE => {
- let request = StoreConsumerOffset2Request::decode_from(payload)
- .map_err(|_| IggyError::InvalidCommand)?;
- namespace_from_partition(&request.stream_id, &request.topic_id, request.partition_id)?
- }
- DELETE_CONSUMER_OFFSET_2_CODE => {
- let request = DeleteConsumerOffset2Request::decode_from(payload)
- .map_err(|_| IggyError::InvalidCommand)?;
- namespace_from_partition(&request.stream_id, &request.topic_id, request.partition_id)?
- }
- DELETE_SEGMENTS_CODE => {
- let request = DeleteSegmentsRequest::decode_from(payload)
- .map_err(|_| IggyError::InvalidCommand)?;
- namespace_from_partition(
- &request.stream_id,
- &request.topic_id,
- Some(request.partition_id),
- )?
- }
- _ => return Err(IggyError::FeatureUnavailable),
- };
-
- Ok(namespace)
-}
-
-fn namespace_from_partitioning(
- stream_id: &WireIdentifier,
- topic_id: &WireIdentifier,
- partitioning: &WirePartitioning,
-) -> Result<u64, IggyError> {
- let WirePartitioning::PartitionId(partition_id) = partitioning else {
- return Err(IggyError::FeatureUnavailable);
- };
- namespace_from_partition(stream_id, topic_id, Some(*partition_id))
-}
-
-fn namespace_from_partition(
- stream_id: &WireIdentifier,
- topic_id: &WireIdentifier,
- partition_id: Option<u32>,
-) -> Result<u64, IggyError> {
- let partition_id = partition_id.ok_or(IggyError::InvalidIdentifier)?;
- let Some(stream_id) = stream_id.as_u32() else {
- return Ok(0);
- };
- let Some(topic_id) = topic_id.as_u32() else {
- return Ok(0);
- };
- validate_namespace_field(stream_id, MAX_STREAMS)?;
- validate_namespace_field(topic_id, MAX_TOPICS)?;
- validate_namespace_field(partition_id, MAX_PARTITIONS)?;
- Ok(pack_namespace(
- stream_id as usize,
- topic_id as usize,
- partition_id as usize,
- ))
-}
-
-fn validate_namespace_field(value: u32, exclusive_max: usize) -> Result<(), IggyError> {
- let value = usize::try_from(value).map_err(|_| IggyError::InvalidIdentifier)?;
- if value >= exclusive_max {
- return Err(IggyError::InvalidIdentifier);
- }
- Ok(())
-}
-
-fn pack_namespace(stream_id: usize, topic_id: usize, partition_id: usize) -> u64 {
- ((stream_id as u64) & STREAM_MASK) << STREAM_SHIFT
- | ((topic_id as u64) & TOPIC_MASK) << TOPIC_SHIFT
- | ((partition_id as u64) & PARTITION_MASK) << PARTITION_SHIFT
-}
-
#[cfg(test)]
mod tests {
use super::*;
@@ -469,7 +357,6 @@ mod tests {
use iggy_binary_protocol::codes::{
CREATE_STREAM_CODE, GET_STREAM_CODE, LOGOUT_USER_CODE, PING_CODE,
};
- use iggy_binary_protocol::requests::messages::SendMessagesHeader;
use iggy_binary_protocol::requests::streams::CreateStreamRequest;
use iggy_binary_protocol::requests::users::LoginRegisterRequest;
use iggy_binary_protocol::version::IGGY_PROTOCOL_VERSION;
@@ -693,47 +580,51 @@ mod tests {
}
#[test]
- fn namespace_defers_named_identifiers_to_server_resolution() {
- let stream = WireIdentifier::named("stream").unwrap();
- let topic = WireIdentifier::numeric(1);
- let namespace = namespace_from_partition(&stream, &topic, Some(0)).unwrap();
- assert_eq!(namespace, 0);
- }
+ fn unknown_code_encodes_as_non_replicated_and_carries_the_code() {
+ // An extended server may implement codes this SDK build has never heard
+ // of. The registry is not a capability list, so the request must reach
+ // the server rather than fail at encode time.
+ const UNKNOWN_CODE: u32 = 60_000;
+ assert!(
+ iggy_binary_protocol::dispatch::lookup_command(UNKNOWN_CODE).is_none(),
+ "test needs a code absent from COMMAND_TABLE"
+ );
- #[test]
- fn namespace_rejects_out_of_range_fields() {
- let stream = WireIdentifier::numeric(MAX_STREAMS as u32);
- let topic = WireIdentifier::numeric(1);
- let err = namespace_from_partition(&stream, &topic, Some(0)).unwrap_err();
- assert!(matches!(err, IggyError::InvalidIdentifier));
+ let mut session = ConsensusSession::with_client_id(42);
+ session.bind(99);
+ let bytes = encode_contiguous_request(&mut session, UNKNOWN_CODE, &Bytes::new()).unwrap();
+ let header = decode_request_header(&bytes);
- let stream = WireIdentifier::numeric(1);
- let partition_id = u32::try_from(MAX_PARTITIONS).unwrap();
- let err = namespace_from_partition(&stream, &topic, Some(partition_id)).unwrap_err();
- assert!(matches!(err, IggyError::InvalidIdentifier));
+ assert_eq!(header.operation, Operation::NonReplicated);
+ assert_eq!(
+ u32::from_le_bytes(
+ header.reserved[NON_REPLICATED_CODE_RANGE]
+ .try_into()
+ .unwrap()
+ ),
+ UNKNOWN_CODE
+ );
}
#[test]
- fn send_messages_with_numeric_partition_builds_namespace() {
- let header = SendMessagesHeader {
- stream_id: WireIdentifier::numeric(2),
- topic_id: WireIdentifier::numeric(3),
- partitioning: WirePartitioning::PartitionId(4),
- messages_count: 0,
- };
- let mut payload = BytesMut::new();
- payload.put_u32_le(header.metadata_length() as u32);
- header.encode(&mut payload);
-
- let namespace = namespace_for_request(
- SEND_MESSAGES_CODE,
- &payload.freeze(),
- Operation::SendMessages,
- )
- .unwrap();
- assert_eq!((namespace >> STREAM_SHIFT) & STREAM_MASK, 2);
- assert_eq!((namespace >> TOPIC_SHIFT) & TOPIC_MASK, 3);
- assert_eq!((namespace >> PARTITION_SHIFT) & PARTITION_MASK, 4);
+ fn no_replicated_command_ever_resolves_to_non_replicated() {
+ // The safety asymmetry that makes forwarding unknown codes acceptable:
+ // an unknown code is the server's business, but a *known* replicated
+ // command sent as non-replicated would apply on one node only and
+ // silently diverge the replicas. Swept over the whole registry so a
+ // future entry cannot regress it.
+ for meta in iggy_binary_protocol::dispatch::COMMAND_TABLE {
+ if !meta.is_replicated() {
+ continue;
+ }
+ assert_ne!(
+ operation_for_code(meta.code).ok(),
+ Some(Operation::NonReplicated),
+ "replicated command {} ({}) must never encode as NonReplicated",
+ meta.name,
+ meta.code
+ );
+ }
}
#[test]
diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs
index 1937a314b..da0bbdf8c 100644
--- a/core/shard/src/metrics.rs
+++ b/core/shard/src/metrics.rs
@@ -94,16 +94,19 @@ pub mod frame_drop_variant {
/// is the receiver-side equivalent: the frame arrived at the owning shard
/// but the local registry refused it. `MISROUTED` ticks when the pump
/// receives a Consensus frame whose target shard is not `self.id`.
+/// `UNCLASSIFIED` ticks when no routing class claims the frame's
+/// operation, so no shard owns it.
pub mod frame_drop_reason {
pub const FULL: &str = "full";
pub const DISCONNECTED: &str = "disconnected";
pub const UNROUTABLE: &str = "unroutable";
pub const DELIVERY_FAILED: &str = "delivery_failed";
pub const MISROUTED: &str = "misrouted";
+ pub const UNCLASSIFIED: &str = "unclassified";
}
const VARIANT_COUNT: usize = 7;
-const REASON_COUNT: usize = 5;
+const REASON_COUNT: usize = 6;
const VARIANTS: [&str; VARIANT_COUNT] = [
frame_drop_variant::CONSENSUS,
@@ -121,6 +124,7 @@ const REASONS: [&str; REASON_COUNT] = [
frame_drop_reason::UNROUTABLE,
frame_drop_reason::DELIVERY_FAILED,
frame_drop_reason::MISROUTED,
+ frame_drop_reason::UNCLASSIFIED,
];
fn variant_index(s: &str) -> Option<usize> {
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index dc0a18e7d..2c26cb07c 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -158,66 +158,64 @@ where
self.try_send_to_target(target, message, operation);
}
- /// Branch on operation class to pick the right routing rule. Used by
- /// [`Self::dispatch`].
- ///
- /// Three classes:
- /// 1. `is_metadata`: metadata data-plane operation, owned by shard 0.
- /// 2. `is_partition`: partition data-plane operation, owned by the
- /// shard that the `IggyNamespace -> shard_id` table assigns.
- /// 3. `is_vsr_reserved` (`Reserved` / `Register`): VSR control-plane
- /// frame (`StartViewChange`, `DoViewChange`, `StartView`, `Commit`)
- /// or a client `Register` request. The owning consensus group is
- /// identified by `namespace_u64`:
- /// - `METADATA_CONSENSUS_NAMESPACE` -> shard 0.
- /// - packable `IggyNamespace::inner()` -> the shard owning that
- /// partition's consensus group.
+ /// Branch on [`RoutingClass`] to pick the right routing rule. Used by
+ /// [`Self::dispatch`]. An operation no class claims has no owning shard,
+ /// so its frame is dropped with `reason=unclassified` rather than falling
+ /// through to consensus-control routing, which would hash it onto an
+ /// arbitrary shard.
fn route_typed(
&self,
operation: Operation,
namespace_u64: u64,
generic: Message<GenericHeader>,
) {
- if operation.is_metadata() {
- self.try_send_to_target(0, generic, operation);
- return;
- }
- if operation.is_partition() {
- let partition_namespace = IggyNamespace::from_raw(namespace_u64);
- // The shards table is a cache of the deterministic hash
- // assignment (every `InsertOwned`/`InsertRouted` row stores
- // `calculate_shard_assignment`'s result), seeded asynchronously
- // by each shard's reconciler. A miss therefore means "not seeded
- // yet", not "unroutable": fall back to the hash so replication
- // frames arriving during the post-commit convergence window are
- // not dropped (the partition plane re-checks materialisation on
- // the owning shard).
- let target = self
- .shards_table
- .shard_for(partition_namespace)
- .unwrap_or_else(|| {
- tracing::debug!(
- shard = self.id,
- stream = partition_namespace.stream_id(),
- topic = partition_namespace.topic_id(),
- partition = partition_namespace.partition_id(),
- "namespace not in shards_table; routing by hash assignment"
- );
- calculate_shard_assignment(&partition_namespace, self.shard_count)
- });
- self.try_send_to_target(target, generic, operation);
- return;
- }
- debug_assert!(
- operation.is_vsr_reserved(),
- "route_typed: operation {operation:?} fell through unclassified; \
- expected is_metadata / is_partition / is_vsr_reserved"
- );
- if namespace_u64 == METADATA_CONSENSUS_NAMESPACE {
- self.try_send_to_target(0, generic, operation);
+ let Some(class) = RoutingClass::of(operation) else {
+ self.metrics.record_frame_drop(
+ frame_drop_variant::CONSENSUS,
+ frame_drop_reason::UNCLASSIFIED,
+ );
+ tracing::error!(
+ shard = self.id,
+ ?operation,
+ "dispatch: operation belongs to no routing class, message dropped"
+ );
return;
+ };
+ match class {
+ RoutingClass::Metadata => self.try_send_to_target(0, generic, operation),
+ RoutingClass::Partition => {
+ let partition_namespace = IggyNamespace::from_raw(namespace_u64);
+ // The shards table is a cache of the deterministic hash
+ // assignment (every `InsertOwned`/`InsertRouted` row stores
+ // `calculate_shard_assignment`'s result), seeded asynchronously
+ // by each shard's reconciler. A miss therefore means "not seeded
+ // yet", not "unroutable": fall back to the hash so replication
+ // frames arriving during the post-commit convergence window are
+ // not dropped (the partition plane re-checks materialisation on
+ // the owning shard).
+ let target = self
+ .shards_table
+ .shard_for(partition_namespace)
+ .unwrap_or_else(|| {
+ tracing::debug!(
+ shard = self.id,
+ stream = partition_namespace.stream_id(),
+ topic = partition_namespace.topic_id(),
+ partition = partition_namespace.partition_id(),
+ "namespace not in shards_table; routing by hash assignment"
+ );
+ calculate_shard_assignment(&partition_namespace, self.shard_count)
+ });
+ self.try_send_to_target(target, generic, operation);
+ }
+ RoutingClass::ConsensusControl => {
+ if namespace_u64 == METADATA_CONSENSUS_NAMESPACE {
+ self.try_send_to_target(0, generic, operation);
+ } else {
+ self.route_consensus_control(generic, namespace_u64, operation);
+ }
+ }
}
- self.route_consensus_control(generic, namespace_u64, operation);
}
/// Send `message` into `senders[target]`. Honors the `io_uring` reactor
@@ -719,3 +717,75 @@ enum ReplicaHandshakeOutcome {
ReleaseSlot(u64),
ClearDial(u8),
}
+
+/// Which routing rule owns an operation's frames.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum RoutingClass {
+ /// Metadata data-plane operation, owned by shard 0.
+ Metadata,
+ /// Partition data-plane operation, owned by the shard that the
+ /// `IggyNamespace -> shard_id` table assigns.
+ Partition,
+ /// VSR control-plane frame (`StartViewChange`, `DoViewChange`,
+ /// `StartView`, `Commit`) or a client `Register` request. The owning
+ /// consensus group is identified by the frame's namespace:
+ /// `METADATA_CONSENSUS_NAMESPACE` -> shard 0, a packable
+ /// `IggyNamespace::inner()` -> the shard owning that partition's
+ /// consensus group.
+ ConsensusControl,
+}
+
+impl RoutingClass {
+ /// `None` when no class claims `operation`, which leaves its frames
+ /// without an owning shard.
+ const fn of(operation: Operation) -> Option<Self> {
+ if operation.is_metadata() {
+ Some(Self::Metadata)
+ } else if operation.is_partition() {
+ Some(Self::Partition)
+ } else if operation.is_vsr_reserved() {
+ Some(Self::ConsensusControl)
+ } else {
+ None
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn every_routed_operation_resolves_to_its_own_class() {
+ assert_eq!(
+ RoutingClass::of(Operation::CreateStream),
+ Some(RoutingClass::Metadata)
+ );
+ assert_eq!(
+ RoutingClass::of(Operation::CreateTopicWithAssignments),
+ Some(RoutingClass::Metadata)
+ );
+ assert_eq!(
+ RoutingClass::of(Operation::SendMessages),
+ Some(RoutingClass::Partition)
+ );
+ assert_eq!(
+ RoutingClass::of(Operation::Register),
+ Some(RoutingClass::ConsensusControl)
+ );
+ assert_eq!(
+ RoutingClass::of(Operation::Reserved),
+ Some(RoutingClass::ConsensusControl)
+ );
+ }
+
+ #[test]
+ fn an_operation_no_class_claims_resolves_to_none() {
+ // `DeleteSegments` is the one shipped operation no class claims: the
+ // client path resolves it to a `TruncatePartition` before any routing
+ // decision, so it never reaches `route_typed`. It stands in here for
+ // any operation added without a routing class, which `route_typed`
+ // must drop instead of routing as a VSR control frame.
+ assert_eq!(RoutingClass::of(Operation::DeleteSegments), None);
+ }
+}