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/07 12:43:22 UTC
(iggy) 01/02: refactor(consensus): remove namespace from client headers
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch remove_namespace_field
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 9c6e71ffd766c34e1deee9c51a71bf3dc500f972
Author: Grzegorz Koszyk <g....@gmail.com>
AuthorDate: Wed Aug 5 12:13:18 2026 +0200
refactor(consensus): remove namespace from client headers
---
core/binary_protocol/src/consensus/header.rs | 219 +++++++++++++----
core/binary_protocol/src/consensus/mod.rs | 6 +-
core/binary_protocol/src/lib.rs | 5 +-
core/binary_protocol/src/namespace.rs | 36 +--
core/binary_protocol/src/version.rs | 2 +-
core/consensus/src/impls.rs | 160 ++++++------
core/consensus/src/lib.rs | 6 +-
core/consensus/src/observability.rs | 2 +-
core/consensus/src/plane_helpers.rs | 36 ++-
.../tests/cluster/client_table_restart.rs | 5 -
core/integration/tests/sdk/protocol_version.rs | 2 -
core/integration/tests/server/legacy_login_vsr.rs | 2 -
core/metadata/src/impls/metadata.rs | 171 ++++++-------
core/partitions/src/iggy_partition.rs | 38 +--
core/partitions/src/iggy_partitions.rs | 31 +--
core/partitions/src/state_transfer.rs | 6 +-
core/sdk/src/vsr.rs | 270 +--------------------
core/server-ng/src/auth.rs | 8 +-
core/server-ng/src/bootstrap.rs | 2 +-
core/server-ng/src/consumer_group.rs | 10 +-
core/server-ng/src/dispatch.rs | 117 +++++----
core/server-ng/src/dispatch/authz.rs | 8 +-
core/server-ng/src/http/session.rs | 4 +-
core/server-ng/src/http/submit.rs | 8 +-
core/server-ng/src/http/wire.rs | 14 +-
core/server-ng/src/partition_reconciler.rs | 14 +-
core/server-ng/src/pat.rs | 24 +-
core/server-ng/src/responses.rs | 25 +-
core/server-ng/src/session_manager.rs | 4 +-
core/server-ng/src/users.rs | 6 +-
core/server-ng/src/wire.rs | 20 +-
core/server_common/src/consensus_message.rs | 61 ++---
core/server_common/src/send_messages2.rs | 58 ++---
core/server_common/src/sharding/mod.rs | 7 +-
core/server_common/src/sharding/namespace.rs | 12 +-
core/shard/src/lib.rs | 233 +++++++++---------
core/shard/src/router.rs | 36 +--
core/simulator/src/client.rs | 158 ++++++------
core/simulator/src/lib.rs | 30 +--
core/simulator/src/replica.rs | 4 +-
core/simulator/src/workload/auditor.rs | 15 +-
core/simulator/src/workload/mod.rs | 9 +-
core/simulator/src/workload/ops/change_password.rs | 4 +-
.../src/workload/ops/create_consumer_group.rs | 4 +-
.../src/workload/ops/create_partitions.rs | 4 +-
.../workload/ops/create_personal_access_token.rs | 4 +-
core/simulator/src/workload/ops/create_stream.rs | 4 +-
core/simulator/src/workload/ops/create_topic.rs | 4 +-
core/simulator/src/workload/ops/create_user.rs | 4 +-
.../src/workload/ops/delete_consumer_group.rs | 4 +-
.../src/workload/ops/delete_consumer_offset.rs | 4 +-
.../src/workload/ops/delete_consumer_offset_2.rs | 4 +-
.../src/workload/ops/delete_partitions.rs | 4 +-
.../workload/ops/delete_personal_access_token.rs | 4 +-
core/simulator/src/workload/ops/delete_segments.rs | 4 +-
core/simulator/src/workload/ops/delete_stream.rs | 4 +-
core/simulator/src/workload/ops/delete_topic.rs | 4 +-
core/simulator/src/workload/ops/delete_user.rs | 4 +-
core/simulator/src/workload/ops/mod.rs | 4 +-
core/simulator/src/workload/ops/purge_stream.rs | 4 +-
core/simulator/src/workload/ops/purge_topic.rs | 4 +-
core/simulator/src/workload/ops/send_messages.rs | 4 +-
.../src/workload/ops/store_consumer_offset.rs | 4 +-
.../src/workload/ops/store_consumer_offset_2.rs | 4 +-
.../src/workload/ops/update_permissions.rs | 4 +-
core/simulator/src/workload/ops/update_stream.rs | 4 +-
core/simulator/src/workload/ops/update_topic.rs | 4 +-
core/simulator/src/workload/ops/update_user.rs | 4 +-
68 files changed, 931 insertions(+), 1057 deletions(-)
diff --git a/core/binary_protocol/src/consensus/header.rs b/core/binary_protocol/src/consensus/header.rs
index 52b9fb6ee..b5efe8a6c 100644
--- a/core/binary_protocol/src/consensus/header.rs
+++ b/core/binary_protocol/src/consensus/header.rs
@@ -162,7 +162,6 @@ pub struct RequestHeader {
pub request: u64,
pub operation: Operation,
pub operation_padding: [u8; 7],
- pub namespace: u64,
/// Session fence epoch: the commit op of the latest committed `Register`
/// for this `client`. Handed to the client by that register's reply and
/// echoed on every subsequent request.
@@ -183,7 +182,7 @@ pub struct RequestHeader {
/// The submitter's wire value is never trusted. Zero for `Logout`,
/// partition-plane, and server-internal ops.
pub user_id: u32,
- pub reserved: [u8; 52],
+ pub reserved: [u8; 60],
}
const _: () = {
assert!(size_of::<RequestHeader>() == HEADER_SIZE);
@@ -194,9 +193,88 @@ const _: () = {
assert!(
offset_of!(RequestHeader, user_id) == offset_of!(RequestHeader, session) + size_of::<u64>()
);
- assert!(offset_of!(RequestHeader, reserved) + size_of::<[u8; 52]>() == HEADER_SIZE);
+ assert!(offset_of!(RequestHeader, reserved) + size_of::<[u8; 60]>() == HEADER_SIZE);
+};
+
+/// A [`RequestHeader`] AFTER the receiving node resolved its target.
+///
+/// The server-internal shape a client request travels in between shards
+/// (and follower to primary), never on the client wire. `group` carries the
+/// resolved consensus group so the owner shard can route, park, and fence
+/// WITHOUT re-decoding the payload -- clients no longer send any namespace
+/// (it is derived: plane from `operation`, partition group from the body),
+/// so this is where the derivation result lives for the internal hop.
+///
+/// Layout: identical to [`RequestHeader`] with `group` claiming the first
+/// eight reserved bytes, so the in-place `transmute_header` rewrite stays a
+/// same-size copy.
+#[repr(C)]
+#[derive(Debug, Clone, Copy, CheckedBitPattern, NoUninit)]
+pub struct RoutedRequestHeader {
+ pub checksum: u128,
+ pub checksum_body: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub view: u32,
+ pub release: u32,
+ pub command: Command2,
+ pub replica: u8,
+ pub reserved_frame: [u8; 66],
+
+ pub client: u128,
+ pub request_checksum: u128,
+ pub timestamp: u64,
+ pub request: u64,
+ pub operation: Operation,
+ pub operation_padding: [u8; 7],
+ pub session: u64,
+ pub user_id: u32,
+ /// Same offset and meaning as the leading 52 bytes of
+ /// `RequestHeader::reserved` -- this region CARRIES DATA (the
+ /// non-replicated op code range), so `group` must not displace it.
+ pub reserved: [u8; 52],
+ /// The resolved consensus group id (see `binary_protocol::namespace`),
+ /// claiming the TAIL of the client header's reserved area.
+ pub group: u64,
+}
+const _: () = {
+ assert!(size_of::<RoutedRequestHeader>() == HEADER_SIZE);
+ // Every field shared with `RequestHeader` sits at the same offset --
+ // including the data-bearing prefix of `reserved` (the non-replicated
+ // code range) -- so promotion preserves everything a client sent.
+ assert!(offset_of!(RoutedRequestHeader, client) == offset_of!(RequestHeader, client));
+ assert!(offset_of!(RoutedRequestHeader, session) == offset_of!(RequestHeader, session));
+ assert!(offset_of!(RoutedRequestHeader, user_id) == offset_of!(RequestHeader, user_id));
+ assert!(offset_of!(RoutedRequestHeader, reserved) == offset_of!(RequestHeader, reserved));
+ assert!(offset_of!(RoutedRequestHeader, group) + size_of::<u64>() == HEADER_SIZE);
};
+impl Default for RoutedRequestHeader {
+ fn default() -> Self {
+ Self {
+ checksum: 0,
+ checksum_body: 0,
+ cluster: 0,
+ size: 0,
+ view: 0,
+ release: 0,
+ command: Command2::Request,
+ replica: 0,
+ reserved_frame: [0; 66],
+ client: 0,
+ request_checksum: 0,
+ timestamp: 0,
+ request: 0,
+ operation: Operation::Reserved,
+ operation_padding: [0; 7],
+ session: 0,
+ user_id: 0,
+ reserved: [0; 52],
+ group: 0,
+ }
+ }
+}
+
impl Default for RequestHeader {
fn default() -> Self {
Self {
@@ -215,14 +293,76 @@ impl Default for RequestHeader {
request: 0,
operation: Operation::Reserved,
operation_padding: [0; 7],
- namespace: 0,
session: 0,
user_id: 0,
- reserved: [0; 52],
+ reserved: [0; 60],
+ }
+ }
+}
+
+impl RoutedRequestHeader {
+ /// Promote a client-wire [`RequestHeader`] to the server-internal routed
+ /// shape, stamping the resolved consensus `group`. Every shared field is
+ /// copied verbatim; this is the ONLY sanctioned crossing between the two
+ /// layouts (transmute-based reads across them would alias `group` with
+ /// reserved bytes).
+ #[must_use]
+ pub const fn from_request(header: &RequestHeader, group: u64) -> Self {
+ Self {
+ checksum: header.checksum,
+ checksum_body: header.checksum_body,
+ cluster: header.cluster,
+ size: header.size,
+ view: header.view,
+ release: header.release,
+ command: header.command,
+ replica: header.replica,
+ reserved_frame: header.reserved_frame,
+ client: header.client,
+ request_checksum: header.request_checksum,
+ timestamp: header.timestamp,
+ request: header.request,
+ operation: header.operation,
+ operation_padding: header.operation_padding,
+ session: header.session,
+ user_id: header.user_id,
+ reserved: {
+ let mut reserved = [0u8; 52];
+ let mut i = 0;
+ while i < 52 {
+ reserved[i] = header.reserved[i];
+ i += 1;
+ }
+ reserved
+ },
+ group,
}
}
}
+impl ConsensusHeader for RoutedRequestHeader {
+ const COMMAND: Command2 = Command2::Request;
+ fn operation(&self) -> Operation {
+ self.operation
+ }
+ fn command(&self) -> Command2 {
+ self.command
+ }
+ fn size(&self) -> u32 {
+ self.size
+ }
+
+ fn validate(&self) -> Result<(), ConsensusError> {
+ if self.command != Command2::Request {
+ return Err(ConsensusError::InvalidCommand {
+ expected: Command2::Request,
+ found: self.command,
+ });
+ }
+ Ok(())
+ }
+}
+
impl ConsensusHeader for RequestHeader {
const COMMAND: Command2 = Command2::Request;
fn operation(&self) -> Operation {
@@ -318,7 +458,6 @@ pub struct ReplyHeader {
pub request: u64,
pub operation: Operation,
pub operation_padding: [u8; 7],
- pub namespace: u64,
/// Request-level status: 0 = ok; nonzero = the `IggyError` code for a
/// failure decided before commit (e.g. a dispatch-time authorization
/// denial, or the partition primary rejecting a consumer-offset op).
@@ -332,7 +471,7 @@ pub struct ReplyHeader {
/// `user_id` in `RequestHeader` / `PrepareHeader`; no existing field offset
/// moves and `validate` does not inspect it.
pub status: u32,
- pub reserved: [u8; 28],
+ pub reserved: [u8; 36],
}
const _: () = {
assert!(size_of::<ReplyHeader>() == HEADER_SIZE);
@@ -340,10 +479,7 @@ const _: () = {
offset_of!(ReplyHeader, request_checksum)
== offset_of!(ReplyHeader, reserved_frame) + size_of::<[u8; 66]>()
);
- assert!(
- offset_of!(ReplyHeader, status) == offset_of!(ReplyHeader, namespace) + size_of::<u64>()
- );
- assert!(offset_of!(ReplyHeader, reserved) + size_of::<[u8; 28]>() == HEADER_SIZE);
+ assert!(offset_of!(ReplyHeader, reserved) + size_of::<[u8; 36]>() == HEADER_SIZE);
};
impl Default for ReplyHeader {
@@ -367,9 +503,8 @@ impl Default for ReplyHeader {
request: 0,
operation: Operation::Reserved,
operation_padding: [0; 7],
- namespace: 0,
status: 0,
- reserved: [0; 28],
+ reserved: [0; 36],
}
}
}
@@ -655,7 +790,12 @@ pub struct PrepareHeader {
pub request: u64,
pub operation: Operation,
pub operation_padding: [u8; 7],
- pub namespace: u64,
+ /// Consensus group id: which of the node's multiplexed VSR groups this
+ /// frame belongs to. `METADATA_GROUP` (top bit) for the metadata plane,
+ /// otherwise the partition's packed stream-topic-partition key. The
+ /// demux and repair-replay routing key; see `binary_protocol::namespace`
+ /// for the value-space contract.
+ pub group: u64,
/// Acting user id, copied verbatim from the admitted `RequestHeader`; see
/// that field for the stamping contract.
pub user_id: u32,
@@ -668,8 +808,7 @@ const _: () = {
== offset_of!(PrepareHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(
- offset_of!(PrepareHeader, user_id)
- == offset_of!(PrepareHeader, namespace) + size_of::<u64>()
+ offset_of!(PrepareHeader, user_id) == offset_of!(PrepareHeader, group) + size_of::<u64>()
);
assert!(offset_of!(PrepareHeader, reserved) + size_of::<[u8; 28]>() == HEADER_SIZE);
};
@@ -695,7 +834,7 @@ impl Default for PrepareHeader {
request: 0,
operation: Operation::Reserved,
operation_padding: [0; 7],
- namespace: 0,
+ group: 0,
user_id: 0,
reserved: [0; 28],
}
@@ -785,7 +924,7 @@ pub struct PrepareOkHeader {
pub request: u64,
pub operation: Operation,
pub operation_padding: [u8; 7],
- pub namespace: u64,
+ pub group: u64,
pub reserved: [u8; 48],
}
const _: () = {
@@ -817,7 +956,7 @@ impl Default for PrepareOkHeader {
request: 0,
operation: Operation::Reserved,
operation_padding: [0; 7],
- namespace: 0,
+ group: 0,
reserved: [0; 48],
}
}
@@ -866,7 +1005,7 @@ pub struct CommitHeader {
pub timestamp_monotonic: u64,
pub commit: u64,
pub checkpoint_op: u64,
- pub namespace: u64,
+ pub group: u64,
pub reserved: [u8; 80],
}
const _: () = {
@@ -917,13 +1056,13 @@ pub struct StartViewChangeHeader {
pub replica: u8,
pub reserved_frame: [u8; 66],
- pub namespace: u64,
+ pub group: u64,
pub reserved: [u8; 120],
}
const _: () = {
assert!(size_of::<StartViewChangeHeader>() == HEADER_SIZE);
assert!(
- offset_of!(StartViewChangeHeader, namespace)
+ offset_of!(StartViewChangeHeader, group)
== offset_of!(StartViewChangeHeader, reserved_frame) + size_of::<[u8; 66]>()
);
assert!(offset_of!(StartViewChangeHeader, reserved) + size_of::<[u8; 120]>() == HEADER_SIZE);
@@ -975,7 +1114,7 @@ pub struct DoViewChangeHeader {
pub op: u64,
/// Highest committed op.
pub commit: u64,
- pub namespace: u64,
+ pub group: u64,
/// View when status was last normal (key for log selection).
pub log_view: u32,
pub reserved: [u8; 100],
@@ -1047,7 +1186,7 @@ pub struct StartViewHeader {
pub op: u64,
/// max(commit) from all DVCs.
pub commit: u64,
- pub namespace: u64,
+ pub group: u64,
pub reserved: [u8; 88],
/// Sender's incarnation, echoed from the `RequestStartView` this answers so a
/// recovering replica can prove the reply post-dates its restart (see
@@ -1055,7 +1194,7 @@ pub struct StartViewHeader {
/// (a normal view-change completion), which carries no freshness claim.
///
/// Carved from the tail of the former `reserved` region and placed LAST so it
- /// lands 16-aligned with no padding WITHOUT moving `op`/`commit`/`namespace`.
+ /// lands 16-aligned with no padding WITHOUT moving `op`/`commit`/`group`.
/// A peer that predates it sends zeros, decoding as `incarnation == 0`, which
/// the `handle_start_view` guard treats as no claim rather than as a foreign
/// one, so a mixed-version rolling upgrade is wire-compatible: the pre-upgrade
@@ -1064,7 +1203,7 @@ pub struct StartViewHeader {
}
const _: () = {
assert!(size_of::<StartViewHeader>() == HEADER_SIZE);
- // op/commit/namespace keep their pre-incarnation offsets.
+ // op/commit/group keep their pre-incarnation offsets.
assert!(
offset_of!(StartViewHeader, op)
== offset_of!(StartViewHeader, reserved_frame) + size_of::<[u8; 66]>()
@@ -1130,22 +1269,22 @@ pub struct RequestStartViewHeader {
pub replica: u8,
pub reserved_frame: [u8; 66],
- pub namespace: u64,
+ pub group: u64,
pub reserved: [u8; 104],
/// The requester's per-boot incarnation, echoed back in the answering
/// `StartView` so a reply from a previous incarnation is detectable.
///
/// Carved from the tail of the former `reserved` region and placed LAST so it
- /// lands 16-aligned with no padding WITHOUT moving `namespace`. A peer that
+ /// lands 16-aligned with no padding WITHOUT moving `group`. A peer that
/// predates it sends zeros, decoding as `incarnation == 0`, so a mixed-version
/// rolling upgrade is wire-compatible.
pub incarnation: u128,
}
const _: () = {
assert!(size_of::<RequestStartViewHeader>() == HEADER_SIZE);
- // namespace keeps its pre-incarnation offset.
+ // group keeps its pre-incarnation offset.
assert!(
- offset_of!(RequestStartViewHeader, namespace)
+ offset_of!(RequestStartViewHeader, group)
== offset_of!(RequestStartViewHeader, reserved_frame) + size_of::<[u8; 66]>()
);
// `incarnation` is last and 16-aligned, so the struct has no padding.
@@ -1186,7 +1325,7 @@ impl ConsensusHeader for RequestStartViewHeader {
/// Recovering/holed replica -> a Normal peer: request a repair stream.
///
/// Sent to the primary first. Asks for the journaled prepares in
-/// `[from_op, to_op]` for `namespace`. Header-only. The peer answers with
+/// `[from_op, to_op]` for `group`. Header-only. The peer answers with
/// `RepairPrepare` frames in op order, terminated by `RepairDone` or
/// `RangeEvicted`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
@@ -1205,7 +1344,7 @@ pub struct RequestPreparesHeader {
pub nonce: u128,
pub from_op: u64,
pub to_op: u64,
- pub namespace: u64,
+ pub group: u64,
pub reserved: [u8; 88],
}
const _: () = {
@@ -1268,7 +1407,7 @@ pub struct RepairRangeReplyHeader {
pub nonce: u128,
/// `RepairDone`: last op served. `RangeEvicted`: oldest retained op.
pub op: u64,
- pub namespace: u64,
+ pub group: u64,
pub reserved: [u8; 96],
}
const _: () = {
@@ -1338,7 +1477,7 @@ pub struct RequestStateTransferHeader {
pub reserved_frame: [u8; 66],
pub nonce: u128,
- pub namespace: u64,
+ pub group: u64,
pub reserved: [u8; 104],
}
const _: () = {
@@ -1413,7 +1552,7 @@ pub struct StateTransferTargetHeader {
/// Serving primary's applied frontier (`commit_min`) when the descriptor
/// was built. The receiver's tail repair targets past this.
pub commit_op: u64,
- pub namespace: u64,
+ pub group: u64,
pub available: u8,
pub reserved: [u8; 95],
}
@@ -1487,7 +1626,7 @@ pub struct RequestStateChunkHeader {
pub nonce: u128,
pub offset: u64,
- pub namespace: u64,
+ pub group: u64,
pub len: u32,
/// Index into the offered state manifest. Range-checked by the serving
/// handler against the cached offer (the header cannot know the count).
@@ -1562,7 +1701,7 @@ pub struct StateChunkHeader {
pub nonce: u128,
pub offset: u64,
- pub namespace: u64,
+ pub group: u64,
/// Index into the offered state manifest. Range-checked by the receiving
/// handler against its accepted manifest.
pub artifact: u32,
@@ -1760,11 +1899,7 @@ mod tests {
use std::mem::offset_of;
assert_eq!(size_of::<ReplyHeader>(), HEADER_SIZE);
assert_eq!(
- offset_of!(ReplyHeader, status),
- offset_of!(ReplyHeader, namespace) + size_of::<u64>()
- );
- assert_eq!(
- offset_of!(ReplyHeader, reserved) + size_of::<[u8; 28]>(),
+ offset_of!(ReplyHeader, reserved) + size_of::<[u8; 36]>(),
HEADER_SIZE
);
}
diff --git a/core/binary_protocol/src/consensus/mod.rs b/core/binary_protocol/src/consensus/mod.rs
index 83bf5ae61..d49badc57 100644
--- a/core/binary_protocol/src/consensus/mod.rs
+++ b/core/binary_protocol/src/consensus/mod.rs
@@ -48,9 +48,9 @@ pub use header::{
CommitHeader, ConsensusHeader, DoViewChangeHeader, EvictionHeader, EvictionReason,
GenericHeader, HEADER_SIZE, PrepareHeader, PrepareOkHeader, RESERVED_COMMAND_LEN,
RepairPrepareHeader, RepairRangeReplyHeader, ReplyHeader, RequestHeader, RequestPreparesHeader,
- RequestStartViewHeader, RequestStateChunkHeader, RequestStateTransferHeader, SIZE_FIELD_OFFSET,
- StartViewChangeHeader, StartViewHeader, StateChunkHeader, StateTransferTargetHeader,
- read_size_field,
+ RequestStartViewHeader, RequestStateChunkHeader, RequestStateTransferHeader,
+ RoutedRequestHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader, StartViewHeader,
+ StateChunkHeader, StateTransferTargetHeader, read_size_field,
};
pub use operation::Operation;
pub use reply_result::{RESULT_COUNT_LEN, RESULT_ENTRY_LEN, result_code, result_section_len};
diff --git a/core/binary_protocol/src/lib.rs b/core/binary_protocol/src/lib.rs
index 1f4baf3fb..ece340ae3 100644
--- a/core/binary_protocol/src/lib.rs
+++ b/core/binary_protocol/src/lib.rs
@@ -75,8 +75,9 @@ pub use consensus::{
EvictionReason, GenericHeader, HEADER_SIZE, Operation, PrepareHeader, PrepareOkHeader,
RESERVED_COMMAND_LEN, RepairPrepareHeader, RepairRangeReplyHeader, ReplyHeader, RequestHeader,
RequestPreparesHeader, RequestStartViewHeader, RequestStateChunkHeader,
- RequestStateTransferHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader, StartViewHeader,
- StateChunkHeader, StateTransferTargetHeader, read_size_field, result_code, result_section_len,
+ RequestStateTransferHeader, RoutedRequestHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader,
+ StartViewHeader, StateChunkHeader, StateTransferTargetHeader, read_size_field, result_code,
+ result_section_len,
};
pub use dispatch::{COMMAND_TABLE, CommandMeta, lookup_by_operation, lookup_command};
pub use error::WireError;
diff --git a/core/binary_protocol/src/namespace.rs b/core/binary_protocol/src/namespace.rs
index 454e81461..88dbdfa93 100644
--- a/core/binary_protocol/src/namespace.rs
+++ b/core/binary_protocol/src/namespace.rs
@@ -15,14 +15,16 @@
// specific language governing permissions and limitations
// under the License.
-//! Wire-format namespace routing constants.
+//! Consensus group-id packing constants.
//!
-//! Both the SDK encoder (which writes `RequestHeader.namespace`) and the
-//! server-side sharding layer (which hashes it to a shard) must agree on
-//! how stream/topic/partition triples pack into the namespace `u64`. Any
-//! drift between the two silently routes writes to the wrong shard, so the
-//! single source of truth lives here in the wire-format crate that both
-//! sides already depend on.
+//! Clients send no group or namespace at all -- the server derives the
+//! target from the operation and the request payload, stamps it into
+//! `RoutedRequestHeader.group` at the dispatch boundary, and every internal
+//! layer (sharding hash, consensus demux, repair replay) routes on that
+//! stamped value. How stream/topic/partition triples pack into the `u64`
+//! is therefore a server-side agreement between the resolver and the
+//! sharding layer; the single source of truth lives here in the wire-format
+//! crate both already depend on.
pub const MAX_STREAMS: usize = 4096;
pub const MAX_TOPICS: usize = 4096;
@@ -61,20 +63,26 @@ pub const PACKED_NAMESPACE_BITS: u32 = STREAM_BITS + TOPIC_BITS + PARTITION_BITS
/// Equivalent to `(1 << PACKED_NAMESPACE_BITS) - 1`.
pub const PACKED_NAMESPACE_MAX: u64 = (1u64 << PACKED_NAMESPACE_BITS) - 1;
-/// Reserved consensus-namespace identifier for the cluster's metadata replica.
+/// Reserved consensus GROUP id for the cluster's metadata plane.
///
-/// The packed layout uses only bits `0..PACKED_NAMESPACE_BITS`, so the top
-/// bit is unreachable from any packed namespace value. Routers distinguish
-/// metadata's single global consensus group from per-partition consensus
-/// groups by value alone.
-pub const METADATA_CONSENSUS_NAMESPACE: u64 = 1u64 << 63;
+/// The group-id space is not a free namespace: values inside the packed
+/// range are partition groups (the packed stream-topic-partition key), the
+/// top bit is the control plane, and 0 is "unset", legal only on client
+/// request headers. The packed layout uses only bits
+/// `0..PACKED_NAMESPACE_BITS` (compile-asserted below), so the top bit is
+/// unreachable from any packed value and routers distinguish metadata's
+/// single global consensus group from per-partition groups by value alone.
+/// Reserving the BOTTOM of the range instead (Redpanda's raft group 0)
+/// only works for allocated ids; ours are derived, and packed 0 is the
+/// legal partition `(0, 0, 0)`.
+pub const METADATA_GROUP: u64 = 1u64 << 63;
// Compile-time invariants. Bumping `MAX_STREAMS`/`MAX_TOPICS`/`MAX_PARTITIONS`
// past the values here would silently collapse the sentinel-above-packed-range
// guarantee and route writes to the wrong shard; the assertions guard against
// that in every build (release included), not only under `cargo test`.
const _: () = {
- assert!(METADATA_CONSENSUS_NAMESPACE > PACKED_NAMESPACE_MAX);
+ assert!(METADATA_GROUP > PACKED_NAMESPACE_MAX);
assert!(PACKED_NAMESPACE_BITS == STREAM_BITS + TOPIC_BITS + PARTITION_BITS);
assert!(PACKED_NAMESPACE_MAX == (1u64 << PACKED_NAMESPACE_BITS) - 1);
};
diff --git a/core/binary_protocol/src/version.rs b/core/binary_protocol/src/version.rs
index 1eaba07c5..9e7916c41 100644
--- a/core/binary_protocol/src/version.rs
+++ b/core/binary_protocol/src/version.rs
@@ -63,7 +63,7 @@
//! `ClientVersionInfo` is the leading bytes of the login-register request
//! *body*, which itself rides inside a 256-byte VSR `RequestHeader` (see
//! `consensus::header`): `command` = `Command2::Request`, `operation` =
-//! `Operation::Register`, `namespace` = `METADATA_CONSENSUS_NAMESPACE`,
+//! `Operation::Register`, `namespace` = `METADATA_GROUP`,
//! client id in `RequestHeader.client`. A foreign SDK emits that header,
//! then the body starting with this prefix, to reach the gate.
//!
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index 17c3c1fa2..c8379099b 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -28,14 +28,15 @@ use bit_set::BitSet;
use clock::{Clock, IggySystemClock};
use iggy_binary_protocol::{
Command2, ConsensusHeader, DoViewChangeHeader, GenericHeader, PrepareHeader, PrepareOkHeader,
- ReplyHeader, RequestHeader, RequestStartViewHeader, StartViewChangeHeader, StartViewHeader,
+ ReplyHeader, RequestStartViewHeader, RoutedRequestHeader, StartViewChangeHeader,
+ StartViewHeader,
};
use iggy_common::IggyTimestamp;
use iggy_common::calculate_checksum;
use message_bus::IggyMessageBus;
use message_bus::MessageBus;
use server_common::Message;
-use server_common::sharding::{IggyNamespace, METADATA_CONSENSUS_NAMESPACE};
+use server_common::sharding::{IggyNamespace, METADATA_GROUP};
use std::cell::{Cell, RefCell};
use std::collections::VecDeque;
use std::rc::Rc;
@@ -242,7 +243,7 @@ impl PipelineEntry {
/// Accepted request waiting in `request_queue` for a prepare slot.
#[derive(Debug)]
pub struct RequestEntry {
- pub message: Message<RequestHeader>,
+ pub message: Message<RoutedRequestHeader>,
// TODO: populate from monotonic clock at push, promote to `pub` for
// age-based filtering. Currently `0`; `pub(crate)` blocks sort-on-stub.
#[allow(dead_code)]
@@ -256,7 +257,7 @@ pub struct RequestEntry {
impl RequestEntry {
#[must_use]
- pub const fn new(message: Message<RequestHeader>) -> Self {
+ pub const fn new(message: Message<RoutedRequestHeader>) -> Self {
Self {
message,
received_at: 0,
@@ -271,7 +272,7 @@ impl RequestEntry {
/// instead of being bounced with a transient error.
#[must_use]
pub fn with_subscriber(
- message: Message<RequestHeader>,
+ message: Message<RoutedRequestHeader>,
) -> (Self, Receiver<Message<ReplyHeader>>) {
let (sender, receiver) = oneshot::channel();
let entry = Self {
@@ -723,7 +724,7 @@ pub enum CommitOutcome {
#[derive(Debug, Clone)]
pub enum VsrAction {
/// Send `StartViewChange` to all replicas.
- SendStartViewChange { view: u32, namespace: u64 },
+ SendStartViewChange { view: u32, group: u64 },
/// Send `DoViewChange` to primary.
SendDoViewChange {
view: u32,
@@ -731,13 +732,13 @@ pub enum VsrAction {
log_view: u32,
op: u64,
commit: u64,
- namespace: u64,
+ group: u64,
},
/// Broadcast a `RequestStartView` probe (recovering replica asking for
/// the current view's `StartView`; only that view's primary answers).
/// Stamped with the prober's view so peers can fence stale duplicates
/// out of the probed-primary election path.
- SendRequestStartView { view: u32, namespace: u64 },
+ SendRequestStartView { view: u32, group: u64 },
/// Send `StartView`, as the view's primary.
///
/// `incarnation` echoes the requester's nonce when this answers a
@@ -754,7 +755,7 @@ pub enum VsrAction {
commit: u64,
incarnation: u128,
target: Option<u8>,
- namespace: u64,
+ group: u64,
},
/// Send `PrepareOK` for each op in `[from_op, to_op]` that is present in the WAL.
///
@@ -766,7 +767,7 @@ pub enum VsrAction {
from_op: u64,
to_op: u64,
target: u8,
- namespace: u64,
+ group: u64,
},
/// Retransmit uncommitted prepares from the WAL to replicas that haven't acked.
///
@@ -793,7 +794,7 @@ pub enum VsrAction {
SendCommit {
view: u32,
commit: u64,
- namespace: u64,
+ group: u64,
timestamp_monotonic: u64,
},
}
@@ -829,7 +830,7 @@ where
cluster: u128,
replica: u8,
replica_count: u8,
- namespace: u64,
+ group: u64,
view: Cell<u32>,
@@ -952,7 +953,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
cluster: u128,
replica: u8,
replica_count: u8,
- namespace: u64,
+ group: u64,
message_bus: B,
pipeline: P,
) -> Self {
@@ -960,7 +961,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
cluster,
replica,
replica_count,
- namespace,
+ group,
message_bus,
pipeline,
ConsensusClock::system(),
@@ -977,7 +978,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
cluster: u128,
replica: u8,
replica_count: u8,
- namespace: u64,
+ group: u64,
message_bus: B,
pipeline: P,
clock: ConsensusClock,
@@ -988,25 +989,25 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
);
assert!(replica_count >= 1, "need at least 1 replica");
// Consensus-control routing distinguishes metadata frames from
- // partition frames by namespace value: metadata uses the sentinel,
+ // partition frames by the group id: metadata uses the sentinel,
// partitions use `IggyNamespace::inner()` which lives strictly
- // inside the packed range. A namespace outside both ranges would
+ // inside the packed range. A group outside both ranges would
// route to neither and silently warn-drop on every receiving peer.
debug_assert!(
- namespace == METADATA_CONSENSUS_NAMESPACE || IggyNamespace::is_packable(namespace),
- "VsrConsensus namespace must be METADATA_CONSENSUS_NAMESPACE or a packable \
- IggyNamespace; got {namespace:#x}"
+ group == METADATA_GROUP || IggyNamespace::is_packable(group),
+ "VsrConsensus group must be METADATA_GROUP or a packable \
+ IggyNamespace; got {group:#x}"
);
// TODO: Verify that XOR-based seeding provides sufficient jitter diversity
// across groups. Consider using a proper hash (e.g., Murmur3) of
- // (replica_id, namespace) for production.
- let timeout_seed = u128::from(replica) ^ u128::from(namespace);
+ // (replica_id, group) for production.
+ let timeout_seed = u128::from(replica) ^ u128::from(group);
let prepare_queue_max = pipeline.prepare_queue_max();
Self {
cluster,
replica,
replica_count,
- namespace,
+ group,
view: Cell::new(0),
log_view: Cell::new(0),
view_durable: Cell::new(0),
@@ -1431,8 +1432,8 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
}
#[must_use]
- pub const fn namespace(&self) -> u64 {
- self.namespace
+ pub const fn group(&self) -> u64 {
+ self.group
}
#[must_use]
@@ -1693,7 +1694,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
if self.state_transfer_stage.get() != StateTransferStage::Idle {
tracing::info!(
replica = self.replica,
- namespace_raw = self.namespace,
+ namespace_raw = self.group,
"view probe exhausted; abandoning state transfer (cluster bootstrap)"
);
self.set_state_transfer_stage(StateTransferStage::Idle);
@@ -1708,7 +1709,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
.reset(TimeoutKind::RequestStartViewMessage);
actions.push(VsrAction::SendRequestStartView {
view: self.view.get(),
- namespace: self.namespace,
+ group: self.group,
});
}
}
@@ -1718,7 +1719,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
.reset(TimeoutKind::RequestStartViewMessage);
actions.push(VsrAction::SendRequestStartView {
view: self.view.get(),
- namespace: self.namespace,
+ group: self.group,
});
}
_ => {
@@ -1782,7 +1783,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
if self.observed_newer_view.get() > self.view.get() {
tracing::info!(
replica = self.replica,
- namespace_raw = self.namespace,
+ namespace_raw = self.group,
view = self.view.get(),
observed_newer_view = self.observed_newer_view.get(),
"heartbeat timed out behind a newer view; probing to catch up"
@@ -1827,7 +1828,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
let action = VsrAction::SendStartViewChange {
view: new_view,
- namespace: self.namespace,
+ group: self.group,
};
emit_sim_event(
SimEventKind::ControlMessageScheduled,
@@ -1851,7 +1852,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
let action = VsrAction::SendStartViewChange {
view: self.view.get(),
- namespace: self.namespace,
+ group: self.group,
};
emit_sim_event(
SimEventKind::ControlMessageScheduled,
@@ -1890,7 +1891,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
op: current_op,
// commit_max clamped to op: see `handle_start_view_change`.
commit: self.commit_max.get().min(current_op),
- namespace: self.namespace,
+ group: self.group,
};
emit_sim_event(
SimEventKind::ControlMessageScheduled,
@@ -1935,7 +1936,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
let action = VsrAction::SendStartViewChange {
view: next_view,
- namespace: self.namespace,
+ group: self.group,
};
emit_sim_event(
SimEventKind::ControlMessageScheduled,
@@ -2044,7 +2045,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
if self.observed_newer_view.get() > self.view.get() {
tracing::info!(
replica = self.replica,
- namespace_raw = self.namespace,
+ namespace_raw = self.group,
view = self.view.get(),
observed_newer_view = self.observed_newer_view.get(),
"stale primary-by-index behind a newer view; probing to catch up"
@@ -2065,7 +2066,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
vec![VsrAction::SendCommit {
view: self.view.get(),
commit: self.commit_min.get(),
- namespace: self.namespace,
+ group: self.group,
timestamp_monotonic: ts,
}]
}
@@ -2083,10 +2084,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
plane: PlaneKind,
header: &StartViewChangeHeader,
) -> Vec<VsrAction> {
- assert_eq!(
- header.namespace, self.namespace,
- "SVC routed to wrong group"
- );
+ assert_eq!(header.group, self.group, "SVC routed to wrong group");
// A recovering replica is quorum-invisible: it lost (or cannot trust)
// its durable state, so it must not vote history into existence. The
// election proceeds among the peers; its conclusion reaches this
@@ -2137,7 +2135,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
// Send our own SVC
let action = VsrAction::SendStartViewChange {
view: msg_view,
- namespace: self.namespace,
+ group: self.group,
};
emit_sim_event(
SimEventKind::ControlMessageScheduled,
@@ -2191,7 +2189,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
log_view: self.log_view.get(),
op: current_op,
commit,
- namespace: self.namespace,
+ group: self.group,
};
emit_sim_event(
SimEventKind::ControlMessageScheduled,
@@ -2239,10 +2237,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
plane: PlaneKind,
header: &DoViewChangeHeader,
) -> Vec<VsrAction> {
- assert_eq!(
- header.namespace, self.namespace,
- "DVC routed to wrong group"
- );
+ assert_eq!(header.group, self.group, "DVC routed to wrong group");
// Quorum-invisible while recovering (see handle_start_view_change):
// a recovering replica must not collect DVCs and crown itself.
if self.status.get() == Status::Recovering {
@@ -2294,7 +2289,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
// Send our own SVC
let action = VsrAction::SendStartViewChange {
view: msg_view,
- namespace: self.namespace,
+ group: self.group,
};
emit_sim_event(
SimEventKind::ControlMessageScheduled,
@@ -2370,7 +2365,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
pub fn begin_view_probe(&self) {
tracing::info!(
replica = self.replica,
- namespace_raw = self.namespace,
+ namespace_raw = self.group,
"beginning view probe"
);
self.status.set(Status::Recovering);
@@ -2420,7 +2415,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
);
tracing::info!(
replica = self.replica,
- namespace_raw = self.namespace,
+ namespace_raw = self.group,
?from,
?to,
"state transfer stage"
@@ -2446,7 +2441,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
header: &RequestStartViewHeader,
) -> Vec<VsrAction> {
assert_eq!(
- header.namespace, self.namespace,
+ header.group, self.group,
"RequestStartView routed to wrong group"
);
if self.status.get() != Status::Normal {
@@ -2487,7 +2482,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
commit: self.commit_max.get(),
incarnation: header.incarnation,
target: Some(header.replica),
- namespace: self.namespace,
+ group: self.group,
}]
}
@@ -2535,7 +2530,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
/// Gap: if a backup never received a prepare (lost message),
/// `commit_journal` stops at the gap. Requires message repair.
pub fn handle_start_view(&self, plane: PlaneKind, header: &StartViewHeader) -> Vec<VsrAction> {
- assert_eq!(header.namespace, self.namespace, "SV routed to wrong group");
+ assert_eq!(header.group, self.group, "SV routed to wrong group");
let from_replica = header.replica;
let msg_view = header.view;
let msg_op = header.op;
@@ -2673,7 +2668,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
from_op: msg_commit + 1,
to_op: msg_op,
target: from_replica,
- namespace: self.namespace,
+ group: self.group,
};
emit_sim_event(
SimEventKind::ControlMessageScheduled,
@@ -2699,10 +2694,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
/// # Panics
/// If `header.namespace` does not match this replica's namespace.
pub fn handle_commit(&self, header: &iggy_binary_protocol::CommitHeader) -> CommitOutcome {
- assert_eq!(
- header.namespace, self.namespace,
- "Commit routed to wrong group"
- );
+ assert_eq!(header.group, self.group, "Commit routed to wrong group");
if self.is_primary() {
// A heartbeat from the primary of an OLDER view means that
@@ -2843,7 +2835,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
commit: max_commit,
incarnation: 0,
target: None,
- namespace: self.namespace,
+ group: self.group,
};
emit_sim_event(
SimEventKind::ControlMessageScheduled,
@@ -3022,7 +3014,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
}
}
-impl<B, P> Project<Message<PrepareHeader>, VsrConsensus<B, P>> for Message<RequestHeader>
+impl<B, P> Project<Message<PrepareHeader>, VsrConsensus<B, P>> for Message<RoutedRequestHeader>
where
B: MessageBus,
P: Pipeline<Entry = PipelineEntry>,
@@ -3055,7 +3047,7 @@ where
// sealed region before the entry is journaled. Leaving those prepares at `0`
// is the designed "nothing to verify" sentinel, so a future durable partition
// journal skips verification instead of failing every entry as corrupt.
- let checksum_body = if consensus.namespace == METADATA_CONSENSUS_NAMESPACE {
+ let checksum_body = if consensus.group == METADATA_GROUP {
u128::from(calculate_checksum(
&self.as_slice()[size_of::<PrepareHeader>()..],
))
@@ -3080,11 +3072,11 @@ where
timestamp,
operation: old.operation,
// The GROUP's namespace, never the request's: a client
- // RequestHeader carries namespace 0, and journaling that
+ // RoutedRequestHeader carries namespace 0, and journaling that
// would make the stored prepare route to the wrong plane
// when repair later ships it verbatim (live replication
// masked this; repair replay is what broke).
- namespace: consensus.namespace,
+ group: consensus.group,
checksum_body,
// Copied verbatim: carries the stamped acting user for client
// ops (and the authenticated user on Register), so the in-apply
@@ -3119,7 +3111,7 @@ where
commit: consensus.commit_max.get(),
timestamp: old.timestamp,
operation: old.operation,
- namespace: old.namespace,
+ group: old.group,
// PrepareOk is header-only; the frame is exactly the header, so
// `size` is the header size.
size: std::mem::size_of::<PrepareOkHeader>() as u32,
@@ -3137,7 +3129,7 @@ where
type MessageBus = B;
#[rustfmt::skip] // Scuffed formatter. TODO: Make the naming less ambiguous for `Message`.
type Message<H> = Message<H> where H: ConsensusHeader;
- type RequestHeader = RequestHeader;
+ type RoutedRequestHeader = RoutedRequestHeader;
type ReplicateHeader = PrepareHeader;
type AckHeader = PrepareOkHeader;
@@ -3175,20 +3167,20 @@ mod request_queue_tests {
use super::*;
use iggy_binary_protocol::{Command2, Operation};
- fn make_request(client: u128, request_num: u64) -> Message<RequestHeader> {
- let header_size = std::mem::size_of::<RequestHeader>();
- let mut msg = Message::<RequestHeader>::new(header_size);
- let header = bytemuck::checked::try_from_bytes_mut::<RequestHeader>(
+ fn make_request(client: u128, request_num: u64) -> Message<RoutedRequestHeader> {
+ let header_size = std::mem::size_of::<RoutedRequestHeader>();
+ let mut msg = Message::<RoutedRequestHeader>::new(header_size);
+ let header = bytemuck::checked::try_from_bytes_mut::<RoutedRequestHeader>(
&mut msg.as_mut_slice()[..header_size],
)
.expect("zeroed bytes are valid");
- *header = RequestHeader {
+ *header = RoutedRequestHeader {
command: Command2::Request,
client,
session: 1,
request: request_num,
operation: Operation::SendMessages,
- ..RequestHeader::default()
+ ..RoutedRequestHeader::default()
};
msg
}
@@ -3493,7 +3485,7 @@ mod timestamp_clamp_tests {
1,
0,
1,
- METADATA_CONSENSUS_NAMESPACE,
+ METADATA_GROUP,
NoopBus,
LocalPipeline::new(),
lagging_clock,
@@ -3525,7 +3517,7 @@ mod timestamp_clamp_tests {
1,
0,
1,
- METADATA_CONSENSUS_NAMESPACE,
+ METADATA_GROUP,
NoopBus,
LocalPipeline::new(),
leading_clock,
@@ -3558,7 +3550,7 @@ mod timestamp_clamp_tests {
header.commit = op;
header.replica = replica;
header.incarnation = incarnation;
- header.namespace = METADATA_CONSENSUS_NAMESPACE;
+ header.group = METADATA_GROUP;
header.size = size as u32;
msg
}
@@ -3579,7 +3571,7 @@ mod timestamp_clamp_tests {
1,
0,
3,
- METADATA_CONSENSUS_NAMESPACE,
+ METADATA_GROUP,
NoopBus,
LocalPipeline::new(),
ConsensusClock::system(),
@@ -3645,14 +3637,14 @@ mod timestamp_clamp_tests {
LocalPipeline::new(),
ConsensusClock::new(Rc::new(FixedClock(100_000))),
);
- let header_size = size_of::<RequestHeader>();
+ let header_size = size_of::<RoutedRequestHeader>();
let body = b"produce payload";
- let mut msg = Message::<RequestHeader>::new(header_size + body.len());
+ let mut msg = Message::<RoutedRequestHeader>::new(header_size + body.len());
msg.as_mut_slice()[header_size..].copy_from_slice(body);
- let header = bytemuck::checked::try_from_bytes_mut::<RequestHeader>(
+ let header = bytemuck::checked::try_from_bytes_mut::<RoutedRequestHeader>(
&mut msg.as_mut_slice()[..header_size],
)
- .expect("zeroed bytes are a valid RequestHeader");
+ .expect("zeroed bytes are a valid RoutedRequestHeader");
header.command = Command2::Request;
header.client = 1;
header.request = 1;
@@ -3662,7 +3654,7 @@ mod timestamp_clamp_tests {
};
assert_ne!(
- seal(METADATA_CONSENSUS_NAMESPACE),
+ seal(METADATA_GROUP),
0,
"a metadata prepare must be sealed: the WAL scan verifies it after a crash"
);
@@ -3691,7 +3683,7 @@ mod timestamp_clamp_tests {
1,
0,
3,
- METADATA_CONSENSUS_NAMESPACE,
+ METADATA_GROUP,
NoopBus,
LocalPipeline::new(),
ConsensusClock::system(),
@@ -3739,14 +3731,8 @@ mod timestamp_clamp_tests {
/// pins the predicate the dispatch sites and the debug tripwire both read.
#[test]
fn given_view_change_when_needs_superblock_persist_should_track_durability() {
- let mut consensus = VsrConsensus::new(
- 1,
- 0,
- 3,
- METADATA_CONSENSUS_NAMESPACE,
- NoopBus,
- LocalPipeline::new(),
- );
+ let mut consensus =
+ VsrConsensus::new(1, 0, 3, METADATA_GROUP, NoopBus, LocalPipeline::new());
assert!(
!consensus.needs_superblock_persist(),
"fresh replica: view == view_durable == 0"
diff --git a/core/consensus/src/lib.rs b/core/consensus/src/lib.rs
index 10b0425c8..424e253df 100644
--- a/core/consensus/src/lib.rs
+++ b/core/consensus/src/lib.rs
@@ -27,7 +27,7 @@ pub trait Project<T, C: Consensus> {
pub trait Pipeline {
type Entry;
/// Accepted-but-not-yet-prepared client request. For `LocalPipeline`,
- /// `RequestEntry` wrapping `Message<RequestHeader>`.
+ /// `RequestEntry` wrapping `Message<RoutedRequestHeader>`.
type Request;
fn push(&mut self, entry: Self::Entry);
@@ -93,7 +93,7 @@ pub trait Pipeline {
}
}
-pub type RequestMessage<C> = <C as Consensus>::Message<<C as Consensus>::RequestHeader>;
+pub type RequestMessage<C> = <C as Consensus>::Message<<C as Consensus>::RoutedRequestHeader>;
pub type ReplicateMessage<C> = <C as Consensus>::Message<<C as Consensus>::ReplicateHeader>;
pub type AckMessage<C> = <C as Consensus>::Message<<C as Consensus>::AckHeader>;
@@ -102,7 +102,7 @@ pub trait Consensus: Sized {
#[rustfmt::skip] // Scuffed formatter.
type Message<H>: ConsensusMessage<H> where H: ConsensusHeader;
- type RequestHeader: ConsensusHeader;
+ type RoutedRequestHeader: ConsensusHeader;
type ReplicateHeader: ConsensusHeader;
type AckHeader: ConsensusHeader;
diff --git a/core/consensus/src/observability.rs b/core/consensus/src/observability.rs
index 4cd041bc4..e5c58326b 100644
--- a/core/consensus/src/observability.rs
+++ b/core/consensus/src/observability.rs
@@ -249,7 +249,7 @@ impl ReplicaLogContext {
plane,
cluster_id: consensus.cluster(),
replica_id: consensus.replica(),
- namespace: NamespaceLogContext::from_raw(plane, consensus.namespace()),
+ namespace: NamespaceLogContext::from_raw(plane, consensus.group()),
view: consensus.view(),
log_view: consensus.log_view(),
commit: consensus.commit_max(),
diff --git a/core/consensus/src/plane_helpers.rs b/core/consensus/src/plane_helpers.rs
index 4515f2ba7..1a677eeac 100644
--- a/core/consensus/src/plane_helpers.rs
+++ b/core/consensus/src/plane_helpers.rs
@@ -19,7 +19,9 @@ use crate::{
Consensus, IgnoreReason, Pipeline, PipelineEntry, PlaneKind, PrepareOkOutcome, Sequencer,
Status, VsrConsensus,
};
-use iggy_binary_protocol::{Command2, PrepareHeader, PrepareOkHeader, ReplyHeader, RequestHeader};
+use iggy_binary_protocol::{
+ Command2, PrepareHeader, PrepareOkHeader, ReplyHeader, RoutedRequestHeader,
+};
use message_bus::{MessageBus, SendError};
use server_common::{Message, iobuf::Owned};
use std::ops::AsyncFnOnce;
@@ -349,7 +351,6 @@ where
timestamp: prepare_header.timestamp,
request: prepare_header.request,
operation: prepare_header.operation,
- namespace: prepare_header.namespace,
..Default::default()
};
// `BytesMut` makes no alignment guarantee, so never cast into it.
@@ -384,7 +385,7 @@ where
#[must_use]
#[allow(clippy::cast_possible_truncation)]
pub fn build_result_rejection_reply(
- request_header: &RequestHeader,
+ request_header: &RoutedRequestHeader,
commit: u64,
code: u32,
) -> Message<ReplyHeader> {
@@ -412,7 +413,6 @@ pub fn build_result_rejection_reply(
timestamp: request_header.timestamp,
request: request_header.request,
operation: request_header.operation,
- namespace: request_header.namespace,
..Default::default()
};
buffer[..header_size].copy_from_slice(bytemuck::bytes_of(&header));
@@ -436,7 +436,7 @@ pub fn build_result_rejection_reply(
#[allow(clippy::needless_pass_by_value, clippy::cast_possible_truncation)]
pub fn build_reply_from_request<B, P>(
consensus: &VsrConsensus<B, P>,
- request_header: &RequestHeader,
+ request_header: &RoutedRequestHeader,
body: bytes::Bytes,
) -> Message<ReplyHeader>
where
@@ -466,7 +466,6 @@ where
timestamp: request_header.timestamp,
request: request_header.request,
operation: request_header.operation,
- namespace: request_header.namespace,
..Default::default()
};
buffer[..header_size].copy_from_slice(bytemuck::bytes_of(&header));
@@ -492,7 +491,7 @@ where
/// If the constructed message buffer is not valid.
pub fn build_deny_reply_from_request<B, P>(
consensus: &VsrConsensus<B, P>,
- request_header: &RequestHeader,
+ request_header: &RoutedRequestHeader,
status: u32,
) -> Message<ReplyHeader>
where
@@ -524,7 +523,7 @@ where
#[must_use]
#[allow(clippy::cast_possible_truncation)]
pub fn build_deny_reply_from_request_header(
- request_header: &RequestHeader,
+ request_header: &RoutedRequestHeader,
status: u32,
) -> Message<ReplyHeader> {
let header_size = std::mem::size_of::<ReplyHeader>();
@@ -543,7 +542,6 @@ pub fn build_deny_reply_from_request_header(
timestamp: request_header.timestamp,
request: request_header.request,
operation: request_header.operation,
- namespace: request_header.namespace,
..Default::default()
};
buffer[..header_size].copy_from_slice(bytemuck::bytes_of(&header));
@@ -620,7 +618,7 @@ pub async fn send_prepare_ok<B, P>(
prepare_checksum: header.checksum,
request: header.request,
operation: header.operation,
- namespace: header.namespace,
+ group: header.group,
size: std::mem::size_of::<PrepareOkHeader>() as u32,
..Default::default()
};
@@ -753,7 +751,7 @@ mod tests {
new.size = std::mem::size_of::<StartViewChangeHeader>() as u32;
new.view = 1;
new.replica = 0;
- new.namespace = 0;
+ new.group = 0;
});
let actions = consensus.handle_start_view_change(PlaneKind::Metadata, svc.header());
@@ -793,7 +791,7 @@ mod tests {
new.size = std::mem::size_of::<StartViewChangeHeader>() as u32;
new.view = 1;
new.replica = 0;
- new.namespace = 0;
+ new.group = 0;
});
let actions = consensus.handle_start_view_change(PlaneKind::Metadata, svc.header());
@@ -824,7 +822,7 @@ mod tests {
reserved_frame: [0; 66],
op: dvc_op,
commit,
- namespace: 0,
+ group: 0,
log_view: 0,
reserved: [0; 100],
};
@@ -915,7 +913,7 @@ mod tests {
command: Command2::StartViewChange,
replica: 1,
reserved_frame: [0; 66],
- namespace: 0,
+ group: 0,
reserved: [0; 120],
};
let _ = consensus.handle_start_view_change(PlaneKind::Metadata, &svc);
@@ -937,7 +935,7 @@ mod tests {
reserved_frame: [0; 66],
op: 0,
commit: 0,
- namespace: 0,
+ group: 0,
log_view: 0,
reserved: [0; 100],
};
@@ -995,7 +993,7 @@ mod tests {
command: Command2::StartViewChange,
replica: 1,
reserved_frame: [0; 66],
- namespace: 0,
+ group: 0,
reserved: [0; 120],
};
let _ = consensus.handle_start_view_change(PlaneKind::Metadata, &svc);
@@ -1013,7 +1011,7 @@ mod tests {
reserved_frame: [0; 66],
op: winner_op,
commit: 0,
- namespace: 0,
+ group: 0,
log_view: 0,
reserved: [0; 100],
};
@@ -1163,12 +1161,11 @@ mod tests {
consensus.init();
consensus.advance_commit_max(4);
- let request = RequestHeader {
+ let request = RoutedRequestHeader {
command: Command2::Request,
operation: Operation::DeleteConsumerOffset2,
client: 42,
request: 7,
- namespace: 9,
..Default::default()
};
let status = 3021;
@@ -1181,7 +1178,6 @@ mod tests {
assert_eq!(header.commit, 4);
assert_eq!(header.client, 42);
assert_eq!(header.request, 7);
- assert_eq!(header.namespace, 9);
assert_eq!(header.operation, Operation::DeleteConsumerOffset2);
assert_eq!(
header.size as usize,
diff --git a/core/integration/tests/cluster/client_table_restart.rs b/core/integration/tests/cluster/client_table_restart.rs
index 4e572eb5c..de026d7e2 100644
--- a/core/integration/tests/cluster/client_table_restart.rs
+++ b/core/integration/tests/cluster/client_table_restart.rs
@@ -80,7 +80,6 @@ use iggy_binary_protocol::consensus::{
Command2, Operation, ReplyHeader, RequestHeader, read_size_field, result_code,
result_section_len,
};
-use iggy_binary_protocol::namespace::METADATA_CONSENSUS_NAMESPACE;
use iggy_binary_protocol::requests::streams::CreateStreamRequest;
use iggy_binary_protocol::requests::users::LoginRegisterRequest;
use iggy_binary_protocol::responses::users::LoginRegisterResponse;
@@ -278,10 +277,6 @@ fn request_header(
client: CLIENT_ID,
session,
request,
- namespace: match operation {
- Operation::Register => METADATA_CONSENSUS_NAMESPACE,
- _ => 0,
- },
..Default::default()
}
}
diff --git a/core/integration/tests/sdk/protocol_version.rs b/core/integration/tests/sdk/protocol_version.rs
index b24db05f2..d6a35c9e7 100644
--- a/core/integration/tests/sdk/protocol_version.rs
+++ b/core/integration/tests/sdk/protocol_version.rs
@@ -27,7 +27,6 @@
use iggy::prelude::*;
use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::consensus::{Command2, Operation, RequestHeader};
-use iggy_binary_protocol::namespace::METADATA_CONSENSUS_NAMESPACE;
use iggy_binary_protocol::requests::users::LoginRegisterRequest;
use iggy_binary_protocol::{
ClientVersionInfo, HEADER_SIZE, IGGY_PROTOCOL_VERSION, IGGY_PROTOCOL_VERSION_MIN, WireName,
@@ -88,7 +87,6 @@ async fn assert_login_evicted(
client: 0xC0FFEE,
session: 0,
request: 0,
- namespace: METADATA_CONSENSUS_NAMESPACE,
..Default::default()
};
diff --git a/core/integration/tests/server/legacy_login_vsr.rs b/core/integration/tests/server/legacy_login_vsr.rs
index 4161a4c3b..d0abcc6e1 100644
--- a/core/integration/tests/server/legacy_login_vsr.rs
+++ b/core/integration/tests/server/legacy_login_vsr.rs
@@ -32,7 +32,6 @@
use iggy_binary_protocol::HEADER_SIZE;
use iggy_binary_protocol::codes::{LOGIN_USER_CODE, LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE};
use iggy_binary_protocol::consensus::{Command2, Operation, RequestHeader};
-use iggy_binary_protocol::namespace::METADATA_CONSENSUS_NAMESPACE;
use integration::harness::TestHarness;
use integration::iggy_harness;
use std::mem::offset_of;
@@ -72,7 +71,6 @@ async fn assert_legacy_login_code_evicted(harness: &TestHarness, code: u32) {
client: 0xC0FFEE,
session: 0,
request: 0,
- namespace: METADATA_CONSENSUS_NAMESPACE,
..Default::default()
};
// A non-replicated command code travels in the first 4 reserved bytes.
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index 01e0916a2..90cce689c 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -44,7 +44,7 @@ use iggy_binary_protocol::requests::topics::CreateTopicWithAssignmentsRequest as
use iggy_binary_protocol::requests::topics::UpdateTopicRequest as WireUpdateTopicRequest;
use iggy_binary_protocol::{
Command2, ConsensusHeader, EvictionReason, GenericHeader, Operation, PrepareHeader,
- PrepareOkHeader, ReplyHeader, RequestHeader, WireDecode, WireEncode, WireName,
+ PrepareOkHeader, ReplyHeader, RoutedRequestHeader, WireDecode, WireEncode, WireName,
};
use iggy_common::IggyError;
use iggy_common::UserId;
@@ -936,7 +936,10 @@ where
Error = iggy_common::IggyError,
>,
{
- async fn on_request(&self, message: <VsrConsensus<B> as Consensus>::Message<RequestHeader>) {
+ async fn on_request(
+ &self,
+ message: <VsrConsensus<B> as Consensus>::Message<RoutedRequestHeader>,
+ ) {
let Some(consensus) =
require_shard_zero(self.consensus.as_ref(), "on_request", "consensus")
else {
@@ -1867,7 +1870,7 @@ where
}
let request = build_register_request_message(consensus, client_id, user_id);
- // Wire path runs `RequestHeader::validate` at network boundary;
+ // Wire path runs `RoutedRequestHeader::validate` at network boundary;
// in-process skips it. debug_assert pins drift.
debug_assert!(
{
@@ -1969,7 +1972,7 @@ where
/// commit.
fn answer_preflight(
consensus: &VsrConsensus<B>,
- request_header: &RequestHeader,
+ request_header: &RoutedRequestHeader,
outcome: PreflightOutcome,
) -> Option<Result<Message<GenericHeader>, MetadataSubmitError>> {
let client_id = request_header.client;
@@ -2282,10 +2285,10 @@ where
// Build the prepare directly so the `client = 0` header skips the
// client-header validation in `prepare_request` / `Project::project`
// (the in-process path `build_prepare_message` documents).
- let header = RequestHeader {
+ let header = RoutedRequestHeader {
client: 0,
- namespace: server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
- ..RequestHeader::default()
+ group: server_common::sharding::METADATA_GROUP,
+ ..RoutedRequestHeader::default()
};
let prepare = build_prepare_message(
consensus,
@@ -2328,7 +2331,7 @@ where
#[allow(clippy::future_not_send)]
pub async fn submit_request_in_process(
&self,
- message: Message<RequestHeader>,
+ message: Message<RoutedRequestHeader>,
) -> Result<Message<GenericHeader>, MetadataSubmitError> {
let request_header = *message.header();
let client_id = request_header.client;
@@ -3204,7 +3207,7 @@ where
#[allow(clippy::too_many_lines)]
fn prepare_request(
&self,
- mut message: Message<RequestHeader>,
+ mut message: Message<RoutedRequestHeader>,
) -> Result<Message<PrepareHeader>, iggy_common::IggyError> {
let consensus = self.consensus.as_ref().unwrap();
let operation = message.header().operation;
@@ -3235,8 +3238,8 @@ where
if let Some(acting_user_id) =
resolve_acting_user_id(operation, client_id, &self.client_table)?
{
- let request_header = bytemuck::checked::from_bytes_mut::<RequestHeader>(
- &mut message.as_mut_slice()[..size_of::<RequestHeader>()],
+ let request_header = bytemuck::checked::from_bytes_mut::<RoutedRequestHeader>(
+ &mut message.as_mut_slice()[..size_of::<RoutedRequestHeader>()],
);
request_header.user_id = acting_user_id;
}
@@ -3249,7 +3252,7 @@ where
// authz gate. The default arm projects the mutated buffer directly and
// is order-independent.
let header = *message.header();
- let body = &message.as_slice()[size_of::<RequestHeader>()..header.size as usize];
+ let body = &message.as_slice()[size_of::<RoutedRequestHeader>()..header.size as usize];
match header.operation {
Operation::CreateTopic => {
@@ -3502,36 +3505,36 @@ where
}
}
-/// In-process Register `Message<RequestHeader>`. Mirrors
+/// In-process Register `Message<RoutedRequestHeader>`. Mirrors
/// `SimClient::register`: `session=0`, `request=0` per
-/// [`RequestHeader::validate`]; empty body.
+/// [`RoutedRequestHeader::validate`]; empty body.
///
/// `cluster` + `view` from `consensus` for self-consistency before
/// `Project::project` overwrites. `release = 0` matches wire today; both
/// paths should switch to `consensus.release()` once
/// `ClientReleaseTooLow/TooHigh` lands.
///
-/// Buffer is `size_of::<RequestHeader>()`; `prepare_request` transmutes into
+/// Buffer is `size_of::<RoutedRequestHeader>()`; `prepare_request` transmutes into
/// `PrepareHeader` (also 256 bytes), no realloc.
fn build_register_request_message<B, P>(
consensus: &VsrConsensus<B, P>,
client_id: u128,
user_id: u32,
-) -> Message<RequestHeader>
+) -> Message<RoutedRequestHeader>
where
B: MessageBus,
P: Pipeline<Entry = PipelineEntry>,
{
- let header_size = size_of::<RequestHeader>();
- let mut msg = Message::<RequestHeader>::new(header_size);
- let header = bytemuck::checked::try_from_bytes_mut::<RequestHeader>(
+ let header_size = size_of::<RoutedRequestHeader>();
+ let mut msg = Message::<RoutedRequestHeader>::new(header_size);
+ let header = bytemuck::checked::try_from_bytes_mut::<RoutedRequestHeader>(
&mut msg.as_mut_slice()[..header_size],
)
- .expect("zeroed bytes are a valid RequestHeader");
- *header = RequestHeader {
+ .expect("zeroed bytes are a valid RoutedRequestHeader");
+ *header = RoutedRequestHeader {
command: Command2::Request,
operation: Operation::Register,
- size: u32::try_from(header_size).expect("RequestHeader size fits u32"),
+ size: u32::try_from(header_size).expect("RoutedRequestHeader size fits u32"),
cluster: consensus.cluster(),
view: consensus.view(),
release: 0,
@@ -3544,8 +3547,8 @@ where
// prepare is re-routed on each peer by namespace; a `0` here would
// hash to a non-zero shard with no metadata consensus and be
// silently dropped (see `shard::router::route_typed`).
- namespace: server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
- ..RequestHeader::default()
+ group: server_common::sharding::METADATA_GROUP,
+ ..RoutedRequestHeader::default()
};
msg
}
@@ -3555,21 +3558,21 @@ fn build_logout_request_message<B, P>(
client_id: u128,
session: u64,
request: u64,
-) -> Message<RequestHeader>
+) -> Message<RoutedRequestHeader>
where
B: MessageBus,
P: Pipeline<Entry = PipelineEntry>,
{
- let header_size = size_of::<RequestHeader>();
- let mut msg = Message::<RequestHeader>::new(header_size);
- let header = bytemuck::checked::try_from_bytes_mut::<RequestHeader>(
+ let header_size = size_of::<RoutedRequestHeader>();
+ let mut msg = Message::<RoutedRequestHeader>::new(header_size);
+ let header = bytemuck::checked::try_from_bytes_mut::<RoutedRequestHeader>(
&mut msg.as_mut_slice()[..header_size],
)
- .expect("zeroed bytes are a valid RequestHeader");
- *header = RequestHeader {
+ .expect("zeroed bytes are a valid RoutedRequestHeader");
+ *header = RoutedRequestHeader {
command: Command2::Request,
operation: Operation::Logout,
- size: u32::try_from(header_size).expect("RequestHeader size fits u32"),
+ size: u32::try_from(header_size).expect("RoutedRequestHeader size fits u32"),
cluster: consensus.cluster(),
view: consensus.view(),
release: 0,
@@ -3577,8 +3580,8 @@ where
session,
request,
// Metadata consensus group (see `build_register_request_message`).
- namespace: server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
- ..RequestHeader::default()
+ group: server_common::sharding::METADATA_GROUP,
+ ..RoutedRequestHeader::default()
};
msg
}
@@ -3588,21 +3591,21 @@ fn build_complete_revocation_request_message<B, P>(
client_id: u128,
request: u64,
body: &[u8],
-) -> Message<RequestHeader>
+) -> Message<RoutedRequestHeader>
where
B: MessageBus,
P: Pipeline<Entry = PipelineEntry>,
{
- let header_size = size_of::<RequestHeader>();
+ let header_size = size_of::<RoutedRequestHeader>();
let total = header_size + body.len();
- let mut msg = Message::<RequestHeader>::new(total);
+ let mut msg = Message::<RoutedRequestHeader>::new(total);
{
let slice = msg.as_mut_slice();
slice[header_size..total].copy_from_slice(body);
let header =
- bytemuck::checked::try_from_bytes_mut::<RequestHeader>(&mut slice[..header_size])
- .expect("zeroed bytes are a valid RequestHeader");
- *header = RequestHeader {
+ bytemuck::checked::try_from_bytes_mut::<RoutedRequestHeader>(&mut slice[..header_size])
+ .expect("zeroed bytes are a valid RoutedRequestHeader");
+ *header = RoutedRequestHeader {
command: Command2::Request,
operation: Operation::CompleteConsumerGroupRevocation,
size: u32::try_from(total).expect("request size fits u32"),
@@ -3614,8 +3617,8 @@ where
// there is no real session (the commit path skips reply-caching).
session: 1,
request,
- namespace: server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
- ..RequestHeader::default()
+ group: server_common::sharding::METADATA_GROUP,
+ ..RoutedRequestHeader::default()
};
}
msg
@@ -3638,14 +3641,14 @@ where
/// a few fixed-width fields, so this cannot happen in practice.
#[must_use]
pub fn build_truncate_partition_client_message(
- template: &RequestHeader,
+ template: &RoutedRequestHeader,
client_id: u128,
session: u64,
stream_id: u32,
topic_id: u32,
partition_id: u32,
up_to_offset: u64,
-) -> Message<RequestHeader> {
+) -> Message<RoutedRequestHeader> {
build_truncate_partition_client_message_with_identifiers(
template,
client_id,
@@ -3669,14 +3672,14 @@ pub fn build_truncate_partition_client_message(
/// a few small fields, so this cannot happen in practice.
#[must_use]
pub fn build_truncate_partition_client_message_with_identifiers(
- template: &RequestHeader,
+ template: &RoutedRequestHeader,
client_id: u128,
session: u64,
stream_id: WireIdentifier,
topic_id: WireIdentifier,
partition_id: u32,
up_to_offset: u64,
-) -> Message<RequestHeader> {
+) -> Message<RoutedRequestHeader> {
let body = TruncatePartitionRequest {
stream_id,
topic_id,
@@ -3684,16 +3687,16 @@ pub fn build_truncate_partition_client_message_with_identifiers(
up_to_offset,
}
.to_bytes();
- let header_size = size_of::<RequestHeader>();
+ let header_size = size_of::<RoutedRequestHeader>();
let total = header_size + body.len();
- let mut msg = Message::<RequestHeader>::new(total);
+ let mut msg = Message::<RoutedRequestHeader>::new(total);
{
let slice = msg.as_mut_slice();
slice[header_size..total].copy_from_slice(&body);
let header =
- bytemuck::checked::try_from_bytes_mut::<RequestHeader>(&mut slice[..header_size])
- .expect("zeroed bytes are a valid RequestHeader");
- *header = RequestHeader {
+ bytemuck::checked::try_from_bytes_mut::<RoutedRequestHeader>(&mut slice[..header_size])
+ .expect("zeroed bytes are a valid RoutedRequestHeader");
+ *header = RoutedRequestHeader {
command: Command2::Request,
operation: Operation::TruncatePartition,
size: u32::try_from(total).expect("request size fits u32"),
@@ -3703,8 +3706,8 @@ pub fn build_truncate_partition_client_message_with_identifiers(
client: client_id,
session,
request: template.request,
- namespace: server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
- ..RequestHeader::default()
+ group: server_common::sharding::METADATA_GROUP,
+ ..RoutedRequestHeader::default()
};
}
msg
@@ -3712,7 +3715,7 @@ pub fn build_truncate_partition_client_message_with_identifiers(
fn build_prepare_message<B, P>(
consensus: &VsrConsensus<B, P>,
- request: &RequestHeader,
+ request: &RoutedRequestHeader,
operation: Operation,
body: &[u8],
) -> Message<PrepareHeader>
@@ -3759,7 +3762,7 @@ where
operation,
// The group's namespace, never the request's: clients send 0, and a
// journaled 0 mis-routes the entry when repair replays it verbatim.
- namespace: consensus.namespace(),
+ group: consensus.group(),
// Carry the acting user id so the in-apply RBAC gate sees the same
// identity on every replica. The default projection copies it (see
// `Project::project`); this helper builds prepares for the ops it
@@ -3790,7 +3793,7 @@ const fn eviction_reason_for_invalid(operation: Operation) -> EvictionReason {
}
/// Resolve the acting user id to stamp into a client op's replicated
-/// `RequestHeader`, so the in-apply RBAC gate (`crate::stm::authz`) reads the
+/// `RoutedRequestHeader`, so the in-apply RBAC gate (`crate::stm::authz`) reads the
/// same identity on every replica (WAL replay has no session table).
///
/// - `Ok(Some(id))`: overwrite the header's `user_id` with the committed
@@ -3867,7 +3870,7 @@ fn log_commit_reply_outcome(outcome: CommitReply, client_id: u128, op: u64) {
/// A cached REJECTION replays untouched: it carries no secret, so serving it
/// is both safe and useful.
fn unreplayable_secret_refusal(
- request_header: &RequestHeader,
+ request_header: &RoutedRequestHeader,
cached: &Frozen<{ server_common::MESSAGE_ALIGN }>,
commit: u64,
client_id: u128,
@@ -4156,7 +4159,7 @@ mod tests {
1,
0,
1,
- server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ server_common::sharding::METADATA_GROUP,
NoopBus,
LocalPipeline::new(),
);
@@ -4263,7 +4266,7 @@ mod tests {
1,
0,
1,
- server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ server_common::sharding::METADATA_GROUP,
NoopBus,
LocalPipeline::new(),
);
@@ -4271,7 +4274,7 @@ mod tests {
IggyMetadata::new(Some(consensus), None, None, None, TestMux::default(), None)
}
- fn create_topic_request(client: u128, wire_user_id: u32) -> Message<RequestHeader> {
+ fn create_topic_request(client: u128, wire_user_id: u32) -> Message<RoutedRequestHeader> {
let body = CreateTopicRequest {
stream_id: WireIdentifier::numeric(1),
partitions_count: 1,
@@ -4282,15 +4285,15 @@ mod tests {
name: WireName::new("t").unwrap(),
}
.to_bytes();
- let header_size = size_of::<RequestHeader>();
+ let header_size = size_of::<RoutedRequestHeader>();
let total = header_size + body.len();
- let mut message = Message::<RequestHeader>::new(total);
+ let mut message = Message::<RoutedRequestHeader>::new(total);
{
let slice = message.as_mut_slice();
slice[header_size..total].copy_from_slice(&body);
let header =
- bytemuck::checked::from_bytes_mut::<RequestHeader>(&mut slice[..header_size]);
- *header = RequestHeader {
+ bytemuck::checked::from_bytes_mut::<RoutedRequestHeader>(&mut slice[..header_size]);
+ *header = RoutedRequestHeader {
command: Command2::Request,
operation: Operation::CreateTopic,
size: u32::try_from(total).unwrap(),
@@ -4298,7 +4301,7 @@ mod tests {
session: 1,
request: 1,
user_id: wire_user_id,
- namespace: server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ group: server_common::sharding::METADATA_GROUP,
..Default::default()
};
}
@@ -4426,7 +4429,7 @@ mod tests {
1,
0,
1,
- server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ server_common::sharding::METADATA_GROUP,
NoopBus,
LocalPipeline::new(),
);
@@ -4521,39 +4524,43 @@ mod tests {
reply
}
- fn pat_create_request(client: u128, request: u64) -> Message<RequestHeader> {
- let header_size = size_of::<RequestHeader>();
- let mut message = Message::<RequestHeader>::new(header_size);
- let header = bytemuck::checked::from_bytes_mut::<RequestHeader>(
+ fn pat_create_request(client: u128, request: u64) -> Message<RoutedRequestHeader> {
+ let header_size = size_of::<RoutedRequestHeader>();
+ let mut message = Message::<RoutedRequestHeader>::new(header_size);
+ let header = bytemuck::checked::from_bytes_mut::<RoutedRequestHeader>(
&mut message.as_mut_slice()[..header_size],
);
- *header = RequestHeader {
+ *header = RoutedRequestHeader {
command: Command2::Request,
operation: Operation::CreatePersonalAccessToken,
size: u32::try_from(header_size).unwrap(),
client,
session: 1,
request,
- namespace: server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ group: server_common::sharding::METADATA_GROUP,
..Default::default()
};
message
}
- fn create_stream_request(client: u128, request: u64, name: &str) -> Message<RequestHeader> {
+ fn create_stream_request(
+ client: u128,
+ request: u64,
+ name: &str,
+ ) -> Message<RoutedRequestHeader> {
let body = iggy_binary_protocol::requests::streams::CreateStreamRequest {
name: WireName::new(name).unwrap(),
}
.to_bytes();
- let header_size = size_of::<RequestHeader>();
+ let header_size = size_of::<RoutedRequestHeader>();
let total = header_size + body.len();
- let mut message = Message::<RequestHeader>::new(total);
+ let mut message = Message::<RoutedRequestHeader>::new(total);
{
let slice = message.as_mut_slice();
slice[header_size..total].copy_from_slice(&body);
let header =
- bytemuck::checked::from_bytes_mut::<RequestHeader>(&mut slice[..header_size]);
- *header = RequestHeader {
+ bytemuck::checked::from_bytes_mut::<RoutedRequestHeader>(&mut slice[..header_size]);
+ *header = RoutedRequestHeader {
command: Command2::Request,
operation: Operation::CreateStream,
size: u32::try_from(total).unwrap(),
@@ -4561,7 +4568,7 @@ mod tests {
session: 1,
request,
user_id: 0,
- namespace: server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ group: server_common::sharding::METADATA_GROUP,
..Default::default()
};
}
@@ -4603,7 +4610,7 @@ mod tests {
1,
0,
1,
- server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ server_common::sharding::METADATA_GROUP,
StallBus::default(),
LocalPipeline::new(),
);
@@ -4753,7 +4760,7 @@ mod tests {
1,
0,
1,
- server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ server_common::sharding::METADATA_GROUP,
NoopBus,
LocalPipeline::new(),
);
@@ -4900,7 +4907,7 @@ mod tests {
1,
0,
1,
- server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ server_common::sharding::METADATA_GROUP,
NoopBus,
LocalPipeline::new(),
);
@@ -5025,7 +5032,7 @@ mod tests {
1,
0,
1,
- server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ server_common::sharding::METADATA_GROUP,
NoopBus,
LocalPipeline::new(),
);
@@ -5161,7 +5168,7 @@ mod tests {
1,
0,
1,
- server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ server_common::sharding::METADATA_GROUP,
NoopBus,
LocalPipeline::new(),
);
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index 95fb0ecbc..c04197605 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -47,7 +47,7 @@ use iggy_binary_protocol::requests::consumer_offsets::{
use iggy_binary_protocol::{
AckLevel, GenericHeader, Operation, PrepareHeader, WireDecode, WireIdentifier,
};
-use iggy_binary_protocol::{PrepareOkHeader, RequestHeader};
+use iggy_binary_protocol::{PrepareOkHeader, RoutedRequestHeader};
use iggy_common::{
ConsumerGroupId, ConsumerGroupOffsets, ConsumerKind, ConsumerOffset, ConsumerOffsets,
IggyByteSize, IggyError, IggyExpiry, IggyTimestamp, PartitionStats, PollingKind,
@@ -224,7 +224,7 @@ where
// are the ones diagnostics actually key on.
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("IggyPartition")
- .field("namespace", &self.consensus.namespace())
+ .field("namespace", &self.consensus.group())
.field("offset", &self.offset)
.field("dirty_offset", &self.dirty_offset)
.field("should_increment_offset", &self.should_increment_offset)
@@ -238,12 +238,12 @@ where
}
/// Post-preflight dispatch in `on_request`: replicate via VSR or take the
-/// `NoAck` leader-local fast path. `RequestHeader` is boxed to avoid the
+/// `NoAck` leader-local fast path. `RoutedRequestHeader` is boxed to avoid the
/// 277-byte inline variant tripping clippy's `large_enum_variant`.
enum Disposition {
Replicate(Message<PrepareHeader>),
NoAck {
- request_header: Box<RequestHeader>,
+ request_header: Box<RoutedRequestHeader>,
kind: ConsumerKind,
consumer_id: u32,
offset: Option<u64>,
@@ -514,7 +514,7 @@ where
target: "iggy.partitions.diag",
plane = "partitions",
replica_id = self.consensus.replica(),
- namespace_raw = self.consensus.namespace(),
+ namespace_raw = self.consensus.group(),
view = state.view,
log_view = state.log_view,
superblock_write_failures = failures,
@@ -892,7 +892,7 @@ where
#[allow(clippy::future_not_send)]
async fn apply_consumer_offset_no_ack(
&self,
- request_header: Box<RequestHeader>,
+ request_header: Box<RoutedRequestHeader>,
kind: ConsumerKind,
consumer_id: u32,
offset: Option<u64>,
@@ -1312,7 +1312,7 @@ where
{
#[must_use]
fn namespace(&self) -> IggyNamespace {
- IggyNamespace::from_raw(self.consensus.namespace())
+ IggyNamespace::from_raw(self.consensus.group())
}
fn partition_dir(&self) -> Option<String> {
@@ -1414,9 +1414,9 @@ where
/// Panics if called when this partition's consensus instance is not the
/// primary, is not in normal status, or is currently syncing.
#[allow(clippy::future_not_send, clippy::too_many_lines)]
- pub async fn on_request(&mut self, message: Message<RequestHeader>) {
+ pub async fn on_request(&mut self, message: Message<RoutedRequestHeader>) {
self.clear_pending_consumer_offset_commits_if_view_changed();
- let namespace = IggyNamespace::from_raw(message.header().namespace);
+ let namespace = IggyNamespace::from_raw(message.header().group);
let client_id = message.header().client;
let request = message.header().request;
@@ -2025,7 +2025,7 @@ where
) -> Result<(), IggyError> {
let header = *message.header();
let replica_id = self.consensus.replica();
- let namespace_raw = self.consensus.namespace();
+ let namespace_raw = self.consensus.group();
match header.operation {
Operation::SendMessages => {
@@ -2436,7 +2436,7 @@ where
send_client_replies: bool,
) {
let replica_id = self.consensus.replica();
- let namespace_raw = self.consensus.namespace();
+ let namespace_raw = self.consensus.group();
let drained_count = drained.len();
if let (Some(first), Some(last)) = (drained.first(), drained.last()) {
debug!(
@@ -2684,13 +2684,13 @@ where
fn parse_consumer_offset_request(
operation: Operation,
- message: &Message<RequestHeader>,
+ message: &Message<RoutedRequestHeader>,
) -> Result<(ConsumerKind, u32, Option<u64>, AckLevel), IggyError> {
let total_size =
usize::try_from(message.header().size).map_err(|_| IggyError::InvalidCommand)?;
let body = message
.as_slice()
- .get(std::mem::size_of::<RequestHeader>()..total_size)
+ .get(std::mem::size_of::<RoutedRequestHeader>()..total_size)
.ok_or(IggyError::InvalidCommand)?;
Self::parse_consumer_offset_payload(operation, body)
}
@@ -2701,7 +2701,7 @@ where
/// so nothing replicates.
async fn send_partition_deny_or_log(
consensus: &VsrConsensus<B>,
- header: &RequestHeader,
+ header: &RoutedRequestHeader,
status: u32,
send_fail_label: &'static str,
) {
@@ -4129,7 +4129,7 @@ mod tests {
client_id: u128,
request_id: u64,
consumer_id: u32,
- ) -> Message<RequestHeader> {
+ ) -> Message<RoutedRequestHeader> {
let body = DeleteConsumerOffset2Request {
consumer: WireConsumer::consumer(WireIdentifier::Numeric(consumer_id)),
stream_id: WireIdentifier::Numeric(1),
@@ -4138,17 +4138,17 @@ mod tests {
ack: AckLevel::Quorum,
}
.to_bytes();
- let header_size = std::mem::size_of::<RequestHeader>();
+ let header_size = std::mem::size_of::<RoutedRequestHeader>();
let total = header_size + body.len();
- let mut message = Message::<RequestHeader>::new(total);
+ let mut message = Message::<RoutedRequestHeader>::new(total);
message.as_mut_slice()[header_size..].copy_from_slice(&body);
- message.transmute_header(|_, header: &mut RequestHeader| {
+ message.transmute_header(|_, header: &mut RoutedRequestHeader| {
header.command = Command2::Request;
header.operation = Operation::DeleteConsumerOffset2;
header.client = client_id;
header.session = 1;
header.request = request_id;
- header.namespace = IggyNamespace::new(1, 1, 0).inner();
+ header.group = IggyNamespace::new(1, 1, 0).inner();
header.size = u32::try_from(total).expect("request size fits u32");
})
}
diff --git a/core/partitions/src/iggy_partitions.rs b/core/partitions/src/iggy_partitions.rs
index a25aed657..c216b615d 100644
--- a/core/partitions/src/iggy_partitions.rs
+++ b/core/partitions/src/iggy_partitions.rs
@@ -23,7 +23,7 @@ use crate::{IggyPartition, Partition, PollingArgs, PollingConsumer};
use ahash::AHashSet;
use consensus::{Consensus, Plane, PlaneIdentity, VsrConsensus};
use iggy_binary_protocol::{
- Command2, ConsensusHeader, Operation, PrepareHeader, PrepareOkHeader, RequestHeader,
+ Command2, ConsensusHeader, Operation, PrepareHeader, PrepareOkHeader, RoutedRequestHeader,
};
use journal::superblock::{PingPongSuperblock, SuperblockStore};
use message_bus::MessageBus;
@@ -352,7 +352,7 @@ where
// by the moved partition's namespace key in O(1); the previous
// linear value-scan turned bulk DeleteStream into O(K²) on the
// pump task, stalling client traffic for ~10k-partition topics.
- let moved_ns = IggyNamespace::from_raw(partitions[idx].consensus().namespace());
+ let moved_ns = IggyNamespace::from_raw(partitions[idx].consensus().group());
let entry = self.namespace_map_mut().get_mut(&moved_ns).expect(
"IggyPartitions invariant: swapped-in partition missing namespace_to_local entry",
);
@@ -496,8 +496,11 @@ where
B: MessageBus,
SB: SuperblockStore,
{
- async fn on_request(&self, message: <VsrConsensus<B> as Consensus>::Message<RequestHeader>) {
- let namespace = IggyNamespace::from_raw(message.header().namespace);
+ async fn on_request(
+ &self,
+ message: <VsrConsensus<B> as Consensus>::Message<RoutedRequestHeader>,
+ ) {
+ let namespace = IggyNamespace::from_raw(message.header().group);
if self.is_tombstoned(&namespace) {
warn!(
target: "iggy.partitions.diag",
@@ -550,20 +553,20 @@ where
}
async fn on_replicate(&self, message: <VsrConsensus<B> as Consensus>::Message<PrepareHeader>) {
- let namespace = IggyNamespace::from_raw(message.header().namespace);
- if self.is_tombstoned(&namespace) {
+ let group = IggyNamespace::from_raw(message.header().group);
+ if self.is_tombstoned(&group) {
warn!(
target: "iggy.partitions.diag",
- namespace_raw = namespace.inner(),
+ namespace_raw = group.inner(),
"dropping prepare: namespace tombstoned"
);
return;
}
- let Some(partition) = self.get_mut_by_ns(&namespace) else {
+ let Some(partition) = self.get_mut_by_ns(&group) else {
warn!(
target: "iggy.partitions.diag",
plane = "partitions",
- namespace_raw = namespace.inner(),
+ namespace_raw = group.inner(),
op = message.header().op,
operation = ?message.header().operation,
"partition not initialized for namespace"
@@ -575,21 +578,21 @@ where
#[allow(clippy::too_many_lines)]
async fn on_ack(&self, message: <VsrConsensus<B> as Consensus>::Message<PrepareOkHeader>) {
- let namespace = IggyNamespace::from_raw(message.header().namespace);
- if self.is_tombstoned(&namespace) {
+ let group = IggyNamespace::from_raw(message.header().group);
+ if self.is_tombstoned(&group) {
warn!(
target: "iggy.partitions.diag",
- namespace_raw = namespace.inner(),
+ namespace_raw = group.inner(),
"dropping prepare-ok: namespace tombstoned"
);
return;
}
let config = self.config.clone();
- let Some(partition) = self.get_mut_by_ns(&namespace) else {
+ let Some(partition) = self.get_mut_by_ns(&group) else {
warn!(
target: "iggy.partitions.diag",
plane = "partitions",
- namespace_raw = namespace.inner(),
+ namespace_raw = group.inner(),
op = message.header().op,
"partition not initialized for namespace"
);
diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs
index 9b84dedc4..d529a973f 100644
--- a/core/partitions/src/state_transfer.rs
+++ b/core/partitions/src/state_transfer.rs
@@ -1455,7 +1455,7 @@ where
// Unlink the old segment chain oldest-first (a crash mid-loop leaves
// the NEWEST suffix, which is contiguous) and drop the in-memory
// vectors in lockstep, exactly as `purge` does.
- let namespace_raw = self.consensus().namespace();
+ let namespace_raw = self.consensus().group();
while let Some((_, mut storage)) = self.log.retire_front() {
let (messages_path, index_path) = storage.segment_and_index_paths();
let _ = storage.shutdown();
@@ -1847,7 +1847,7 @@ where
tracing::error!(
target: "iggy.partitions.diag",
plane = "partitions",
- namespace_raw = self.consensus().namespace(),
+ namespace_raw = self.consensus().group(),
partition_dir,
%error,
"converge sweep cannot list the partition directory"
@@ -1861,7 +1861,7 @@ where
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => {
warn_unlink(
- self.consensus().namespace(),
+ self.consensus().group(),
&path.display().to_string(),
&error,
);
diff --git a/core/sdk/src/vsr.rs b/core/sdk/src/vsr.rs
index 9bc5d1cd1..dff9999a8 100644
--- a/core/sdk/src/vsr.rs
+++ b/core/sdk/src/vsr.rs
@@ -17,27 +17,13 @@
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_common::{IggyError, eviction_reason_to_error};
const NON_REPLICATED_CODE_RANGE: std::ops::Range<usize> = 0..4;
@@ -122,12 +108,11 @@ pub(crate) fn encode_request_header(
}
}
};
- let namespace = namespace_for_request(code, payload, operation)?;
let total_size = HEADER_SIZE
.checked_add(payload.len())
.ok_or(IggyError::InvalidConfiguration)?;
let size = u32::try_from(total_size).map_err(|_| IggyError::InvalidConfiguration)?;
- let mut reserved = [0; 52];
+ let mut reserved = [0; 60];
if operation == Operation::NonReplicated {
reserved[NON_REPLICATED_CODE_RANGE].copy_from_slice(&code.to_le_bytes());
}
@@ -138,7 +123,6 @@ pub(crate) fn encode_request_header(
client: session.client_id(),
request: request_id,
session: session_id,
- namespace,
// Zeroed: the field is "informational" -- the server copies it into
// `ReplyHeader.timestamp` for RTT but nothing else reads it. Paying
// a `clock_gettime` syscall per encoded request (formerly held the
@@ -342,132 +326,11 @@ 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::*;
use crate::session::ConsensusSession;
- 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::codes::GET_STREAM_CODE;
use iggy_binary_protocol::requests::users::LoginRegisterRequest;
use iggy_binary_protocol::version::IGGY_PROTOCOL_VERSION;
use iggy_binary_protocol::{ClientVersionInfo, WireEncode, WireName};
@@ -477,34 +340,6 @@ mod tests {
*bytemuck::checked::try_from_bytes::<RequestHeader>(&bytes[..HEADER_SIZE]).unwrap()
}
- #[test]
- fn register_request_uses_zero_request_and_session() {
- let mut session = ConsensusSession::with_client_id(7);
- let request = LoginRegisterRequest {
- version_info: ClientVersionInfo {
- protocol_version: IGGY_PROTOCOL_VERSION,
- sdk_name: WireName::new("rust-sdk").unwrap(),
- sdk_version: WireName::new("1.0.0").unwrap(),
- },
- username: WireName::new("admin").unwrap(),
- password: SecretString::from("secret"),
- client_context: None,
- };
-
- let bytes =
- encode_contiguous_request(&mut session, LOGIN_REGISTER_CODE, &request.to_bytes())
- .unwrap();
- let header = decode_request_header(&bytes);
-
- assert_eq!(header.operation, Operation::Register);
- assert_eq!(header.request, 0);
- assert_eq!(header.session, 0);
- assert_eq!(header.client, 7);
- // Register is routed to the metadata replica (shard 0). The router's
- // namespace==METADATA short-circuit needs the sentinel, not 0.
- assert_eq!(header.namespace, METADATA_CONSENSUS_NAMESPACE);
- }
-
#[test]
fn second_register_on_bound_session_re_arms_instead_of_panicking() {
let request = LoginRegisterRequest {
@@ -614,61 +449,6 @@ mod tests {
assert_eq!(&out[..], b"abc");
}
- #[test]
- fn replicated_request_increments_request_counter() {
- let mut session = ConsensusSession::with_client_id(42);
- let _ = session.register_request_id();
- session.bind(99);
- let payload = CreateStreamRequest {
- name: WireName::new("stream").unwrap(),
- }
- .to_bytes();
-
- let first = encode_contiguous_request(&mut session, CREATE_STREAM_CODE, &payload).unwrap();
- let second = encode_contiguous_request(&mut session, CREATE_STREAM_CODE, &payload).unwrap();
-
- assert_eq!(decode_request_header(&first).request, 1);
- assert_eq!(decode_request_header(&second).request, 2);
- assert_eq!(decode_request_header(&second).session, 99);
- assert_eq!(decode_request_header(&second).namespace, 0);
- }
-
- #[test]
- fn ping_uses_non_replicated_operation() {
- let mut session = ConsensusSession::with_client_id(42);
- session.bind(99);
- let bytes = encode_contiguous_request(&mut session, PING_CODE, &Bytes::new()).unwrap();
- let header = decode_request_header(&bytes);
-
- assert_eq!(header.operation, Operation::NonReplicated);
- assert_eq!(
- u32::from_le_bytes(
- header.reserved[NON_REPLICATED_CODE_RANGE]
- .try_into()
- .unwrap()
- ),
- PING_CODE
- );
- assert_eq!(header.session, 99);
- assert_eq!(header.namespace, 0);
- }
-
- #[test]
- fn logout_uses_replicated_logout_operation() {
- let mut session = ConsensusSession::with_client_id(42);
- session.bind(99);
- let bytes =
- encode_contiguous_request(&mut session, LOGOUT_USER_CODE, &Bytes::new()).unwrap();
- let header = decode_request_header(&bytes);
-
- assert_eq!(header.operation, Operation::Logout);
- assert_eq!(header.request, 1);
- assert_eq!(header.session, 99);
- // Logout, like Register, is routed to shard 0 via the metadata
- // sentinel rather than namespace 0.
- assert_eq!(header.namespace, METADATA_CONSENSUS_NAMESPACE);
- }
-
#[test]
fn read_only_request_uses_non_replicated_operation() {
let mut session = ConsensusSession::with_client_id(42);
@@ -737,50 +517,6 @@ 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);
- }
-
- #[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 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));
- }
-
- #[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);
- }
-
#[test]
fn metadata_success_reply_strips_result_section_and_returns_payload() {
let mut body = BytesMut::new();
diff --git a/core/server-ng/src/auth.rs b/core/server-ng/src/auth.rs
index 74a38444c..c67d00dd2 100644
--- a/core/server-ng/src/auth.rs
+++ b/core/server-ng/src/auth.rs
@@ -29,7 +29,7 @@ use crate::responses::{build_login_register_reply, current_metadata_commit};
use crate::session_manager::{ClientSdkInfo, SessionManager};
use consensus::{MetadataHandle, build_result_rejection_reply};
use iggy_binary_protocol::PrepareHeader;
-use iggy_binary_protocol::{ClientVersionInfo, EvictionReason, RequestHeader};
+use iggy_binary_protocol::{ClientVersionInfo, EvictionReason, RoutedRequestHeader};
use iggy_common::defaults::{
MAX_PASSWORD_LENGTH, MAX_USERNAME_LENGTH, MIN_PASSWORD_LENGTH, MIN_USERNAME_LENGTH,
};
@@ -183,7 +183,7 @@ pub(crate) async fn complete_login_register<B, MJ, S, SB>(
sessions: &Rc<RefCell<SessionManager>>,
transport_client_id: u128,
vsr_client_id: u128,
- request_header: &RequestHeader,
+ request_header: &RoutedRequestHeader,
user_id: u32,
client_version: &ClientVersionInfo,
) -> Result<(), LoginRegisterError>
@@ -291,7 +291,7 @@ where
pub(crate) async fn surface_login_failure<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
transport_client_id: u128,
- request_header: &RequestHeader,
+ request_header: &RoutedRequestHeader,
error: &LoginRegisterError,
) where
B: ShellBus,
@@ -326,7 +326,7 @@ pub(crate) async fn surface_login_failure<B, MJ, S, SB>(
async fn send_login_transient_reply<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
transport_client_id: u128,
- request_header: &RequestHeader,
+ request_header: &RoutedRequestHeader,
) where
B: ShellBus,
MJ: JournalHandle + 'static,
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index 23ecd9108..c635ccddc 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -1990,7 +1990,7 @@ fn restore_metadata_consensus(
topology.cluster_id,
topology.self_replica_id,
replica_count,
- server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ server_common::sharding::METADATA_GROUP,
bus,
// Request queue keeps the stock 2x ratio over the prepare queue
// (32 -> 64 at defaults): buffered requests are cheap relative to
diff --git a/core/server-ng/src/consumer_group.rs b/core/server-ng/src/consumer_group.rs
index 034713644..fd93e3526 100644
--- a/core/server-ng/src/consumer_group.rs
+++ b/core/server-ng/src/consumer_group.rs
@@ -40,7 +40,7 @@ use iggy_binary_protocol::requests::consumer_offsets::{
DeleteConsumerOffset2Request, DeleteConsumerOffsetRequest, StoreConsumerOffset2Request,
StoreConsumerOffsetRequest,
};
-use iggy_binary_protocol::{KIND_CONSUMER_GROUP, Operation, RequestHeader, WireIdentifier};
+use iggy_binary_protocol::{KIND_CONSUMER_GROUP, Operation, RoutedRequestHeader, WireIdentifier};
use iggy_common::IggyError;
use journal::superblock::SuperblockStore;
use journal::{Journal, JournalHandle};
@@ -63,8 +63,8 @@ use std::rc::Rc;
/// join. Every other operation passes through.
pub(crate) async fn maybe_rewrite_consumer_group_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
- request: Message<RequestHeader>,
-) -> Result<Message<RequestHeader>, IggyError>
+ request: Message<RoutedRequestHeader>,
+) -> Result<Message<RoutedRequestHeader>, IggyError>
where
B: ShellBus,
MJ: JournalHandle + 'static,
@@ -208,8 +208,8 @@ where
#[allow(clippy::cast_possible_truncation)]
pub(crate) fn maybe_rewrite_consumer_offset_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
- request: Message<RequestHeader>,
-) -> Result<Message<RequestHeader>, IggyError>
+ request: Message<RoutedRequestHeader>,
+) -> Result<Message<RoutedRequestHeader>, IggyError>
where
B: ShellBus,
MJ: JournalHandle + 'static,
diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs
index bfc11f2a2..6e18ace6f 100644
--- a/core/server-ng/src/dispatch.rs
+++ b/core/server-ng/src/dispatch.rs
@@ -80,8 +80,8 @@ use iggy_binary_protocol::responses::consumer_groups::SyncConsumerGroupResponse;
use iggy_binary_protocol::responses::system::get_snapshot::GetSnapshotResponse;
use iggy_binary_protocol::{
AckLevel, ClientVersionInfo, Command2, EvictionReason, GenericHeader, HEADER_SIZE,
- KIND_CONSUMER_GROUP, Operation, ProtocolVersion, RequestHeader, WireDecode, WireEncode,
- WireIdentifier, is_protocol_compatible,
+ KIND_CONSUMER_GROUP, Operation, ProtocolVersion, RequestHeader, RoutedRequestHeader,
+ WireDecode, WireEncode, WireIdentifier, is_protocol_compatible,
};
use iggy_common::{IggyError, PollingStrategy, SnapshotCompression, SystemSnapshotType};
use journal::superblock::SuperblockStore;
@@ -394,7 +394,7 @@ fn submit_auto_commit<B, MJ, S, SB>(
fn build_auto_commit_request(
namespace: IggyNamespace,
applied: &AutoCommitApplied,
-) -> Result<Message<RequestHeader>, IggyError> {
+) -> Result<Message<RoutedRequestHeader>, IggyError> {
let request = StoreConsumerOffset2Request {
consumer: WireConsumer {
kind: applied.kind.as_code(),
@@ -407,26 +407,28 @@ fn build_auto_commit_request(
ack: AckLevel::Quorum,
};
let body = request.to_bytes();
- let header_size = std::mem::size_of::<RequestHeader>();
+ let header_size = std::mem::size_of::<RoutedRequestHeader>();
let total_size = header_size + body.len();
let size = u32::try_from(total_size).map_err(|_| IggyError::InvalidConfiguration)?;
- let mut message = Message::<RequestHeader>::new(total_size);
+ let mut message = Message::<RoutedRequestHeader>::new(total_size);
message.as_mut_slice()[header_size..].copy_from_slice(&body);
- Ok(message.transmute_header(|_, header: &mut RequestHeader| {
- *header = RequestHeader {
- command: Command2::Request,
- operation: Operation::StoreConsumerOffset2,
- size,
- client: AUTO_COMMIT_CLIENT_ID,
- // The partition plane is sessionless (no `ClientTable` dedup); a
- // nonzero session + request just satisfy the wire header
- // validation.
- session: 1,
- request: 1,
- namespace: namespace.inner(),
- ..Default::default()
- };
- }))
+ Ok(
+ message.transmute_header(|_, header: &mut RoutedRequestHeader| {
+ *header = RoutedRequestHeader {
+ command: Command2::Request,
+ operation: Operation::StoreConsumerOffset2,
+ size,
+ client: AUTO_COMMIT_CLIENT_ID,
+ // The partition plane is sessionless (no `ClientTable` dedup); a
+ // nonzero session + request just satisfy the wire header
+ // validation.
+ session: 1,
+ request: 1,
+ group: namespace.inner(),
+ ..Default::default()
+ };
+ }),
+ )
}
pub(crate) fn make_deferred_replica_message_handler<B, MJ, S, SB>(
@@ -561,7 +563,7 @@ where
let _ = reply.try_send(commit);
}
shard::MetadataSubmit::ClientRequest { request, reply } => {
- let committed = match request.try_into_typed::<RequestHeader>() {
+ let committed = match request.try_into_typed::<RoutedRequestHeader>() {
Ok(typed) => shard
.plane
.metadata()
@@ -759,6 +761,13 @@ async fn handle_client_request<B, MJ, S, SB>(
return;
}
};
+ // Promote to the server-internal routed shape at the boundary: the
+ // client wire carries no group (it is derived -- plane from `operation`,
+ // partition target from the payload), so it starts unset here and the
+ // resolution sites below stamp it before anything routes on it.
+ let request = request.transmute_header(|header, new_header: &mut RoutedRequestHeader| {
+ *new_header = RoutedRequestHeader::from_request(&header, 0);
+ });
ensure_transport_connection(shard, sessions, transport_client_id);
@@ -891,8 +900,10 @@ async fn handle_client_request<B, MJ, S, SB>(
return;
}
- let request = request.transmute_header(|header, new_header: &mut RequestHeader| {
+ let request = request.transmute_header(|header, new_header: &mut RoutedRequestHeader| {
*new_header = header;
+ // Metadata-plane ops route by operation: stamp the sentinel group.
+ new_header.group = server_common::sharding::METADATA_GROUP;
// `bound` is always Some here (unbound transports early-return above);
// this sets the consensus client id + session for the replicated op.
if let Some((bound_client_id, bound_session)) = bound {
@@ -1044,7 +1055,7 @@ async fn handle_get_personal_access_tokens<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
sessions: &Rc<RefCell<SessionManager>>,
transport_client_id: u128,
- request: &Message<RequestHeader>,
+ request: &Message<RoutedRequestHeader>,
) where
B: ShellBus,
MJ: JournalHandle + 'static,
@@ -1071,7 +1082,7 @@ async fn handle_get_me<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
sessions: &Rc<RefCell<SessionManager>>,
transport_client_id: u128,
- request: &Message<RequestHeader>,
+ request: &Message<RoutedRequestHeader>,
) where
B: ShellBus,
MJ: JournalHandle + 'static,
@@ -1109,7 +1120,7 @@ async fn handle_get_me<B, MJ, S, SB>(
#[allow(clippy::future_not_send)]
pub(crate) async fn dispatch_partition_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
- request: Message<RequestHeader>,
+ request: Message<RoutedRequestHeader>,
vsr_client_id: u128,
bound_session: u64,
transport_client_id: u128,
@@ -1221,9 +1232,9 @@ pub(crate) async fn dispatch_partition_request<B, MJ, S, SB>(
return;
}
};
- let request = request.transmute_header(|header, new_header: &mut RequestHeader| {
+ let request = request.transmute_header(|header, new_header: &mut RoutedRequestHeader| {
*new_header = header;
- new_header.namespace = namespace;
+ new_header.group = namespace;
new_header.client = transport_client_id;
// Header validation requires `session > 0 && request > 0` for
// non-register ops. The partition plane itself is sessionless
@@ -1242,7 +1253,7 @@ async fn handle_non_replicated_request<B, MJ, S, SB>(
sessions: &Rc<RefCell<SessionManager>>,
system_config: &Arc<NgSystemConfig>,
transport_client_id: u128,
- request: Message<RequestHeader>,
+ request: Message<RoutedRequestHeader>,
) where
B: ShellBus,
MJ: JournalHandle + 'static,
@@ -1387,7 +1398,7 @@ async fn handle_default_non_replicated<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
transport_client_id: u128,
code: u32,
- request: &Message<RequestHeader>,
+ request: &Message<RoutedRequestHeader>,
user_id: Option<u32>,
roster: &ClusterRoster,
) where
@@ -1451,7 +1462,7 @@ async fn handle_get_snapshot<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
system_config: &Arc<NgSystemConfig>,
transport_client_id: u128,
- request: &Message<RequestHeader>,
+ request: &Message<RoutedRequestHeader>,
user_id: Option<u32>,
) where
B: ShellBus,
@@ -1531,7 +1542,7 @@ fn decode_get_snapshot(
#[allow(clippy::future_not_send)]
async fn send_non_replicated_bytes<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
- request: &Message<RequestHeader>,
+ request: &Message<RoutedRequestHeader>,
transport_client_id: u128,
bytes: Bytes,
label: &'static str,
@@ -1724,7 +1735,7 @@ async fn evict_stale_client<B, MJ, S, SB>(
async fn handle_poll_messages<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
transport_client_id: u128,
- request: &Message<RequestHeader>,
+ request: &Message<RoutedRequestHeader>,
user_id: Option<u32>,
) where
B: ShellBus,
@@ -1823,7 +1834,7 @@ async fn handle_poll_messages<B, MJ, S, SB>(
async fn handle_get_consumer_offset<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
transport_client_id: u128,
- request: &Message<RequestHeader>,
+ request: &Message<RoutedRequestHeader>,
user_id: Option<u32>,
) where
B: ShellBus,
@@ -1896,7 +1907,7 @@ async fn handle_get_consumer_offset<B, MJ, S, SB>(
async fn handle_sync_consumer_group<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
transport_client_id: u128,
- request: &Message<RequestHeader>,
+ request: &Message<RoutedRequestHeader>,
) where
B: ShellBus,
MJ: JournalHandle + 'static,
@@ -1950,7 +1961,7 @@ async fn handle_sync_consumer_group<B, MJ, S, SB>(
async fn send_empty_partition_reply<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
transport_client_id: u128,
- request_header: &RequestHeader,
+ request_header: &RoutedRequestHeader,
) where
B: ShellBus,
MJ: JournalHandle + 'static,
@@ -2277,7 +2288,7 @@ async fn handle_delete_segments_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
transport_client_id: u128,
bound: Option<(u128, u64)>,
- request: &Message<RequestHeader>,
+ request: &Message<RoutedRequestHeader>,
) where
B: ShellBus,
MJ: JournalHandle + 'static,
@@ -2407,11 +2418,11 @@ async fn handle_delete_segments_request<B, MJ, S, SB>(
#[allow(clippy::cast_possible_truncation)]
pub(crate) async fn resolve_delete_segments_truncate<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
- template: &RequestHeader,
+ template: &RoutedRequestHeader,
client_id: u128,
session: u64,
body: &[u8],
-) -> Result<Message<RequestHeader>, IggyError>
+) -> Result<Message<RoutedRequestHeader>, IggyError>
where
B: ShellBus,
MJ: JournalHandle + 'static,
@@ -2572,7 +2583,7 @@ fn submit_disconnect_logout<B, MJ, S, SB>(
#[allow(clippy::future_not_send)]
pub(crate) async fn submit_client_request_on_owner<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
- request: Message<RequestHeader>,
+ request: Message<RoutedRequestHeader>,
) -> Option<Message<GenericHeader>>
where
B: ShellBus,
@@ -2602,7 +2613,7 @@ async fn handle_logout_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
sessions: &Rc<RefCell<SessionManager>>,
transport_client_id: u128,
- request: Message<RequestHeader>,
+ request: Message<RoutedRequestHeader>,
) where
B: ShellBus,
MJ: JournalHandle + 'static,
@@ -2667,7 +2678,7 @@ async fn handle_login_register_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
sessions: &Rc<RefCell<SessionManager>>,
transport_client_id: u128,
- request: Message<RequestHeader>,
+ request: Message<RoutedRequestHeader>,
) where
B: ShellBus,
MJ: JournalHandle + 'static,
@@ -3016,16 +3027,16 @@ mod tests {
session: u64,
request: u64,
body: &[u8],
- ) -> Message<RequestHeader> {
- let header_size = size_of::<RequestHeader>();
+ ) -> Message<RoutedRequestHeader> {
+ let header_size = size_of::<RoutedRequestHeader>();
let total = header_size + body.len();
- let mut message = Message::<RequestHeader>::new(total);
+ let mut message = Message::<RoutedRequestHeader>::new(total);
{
let slice = message.as_mut_slice();
slice[header_size..total].copy_from_slice(body);
let header =
- bytemuck::checked::from_bytes_mut::<RequestHeader>(&mut slice[..header_size]);
- *header = RequestHeader {
+ bytemuck::checked::from_bytes_mut::<RoutedRequestHeader>(&mut slice[..header_size]);
+ *header = RoutedRequestHeader {
command: Command2::Request,
operation,
size: u32::try_from(total).expect("test request fits u32"),
@@ -3033,7 +3044,7 @@ mod tests {
session,
request,
user_id: 0,
- namespace: server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ group: server_common::sharding::METADATA_GROUP,
..Default::default()
};
}
@@ -3067,7 +3078,7 @@ mod tests {
request,
user_id: 0,
checksum: 42,
- namespace: server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ group: server_common::sharding::METADATA_GROUP,
..Default::default()
};
}
@@ -3116,7 +3127,7 @@ mod tests {
1,
0,
1,
- server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ server_common::sharding::METADATA_GROUP,
bus.clone(),
LocalPipeline::new(),
);
@@ -3393,9 +3404,9 @@ mod tests {
shard.plane.partitions().tombstone(namespace);
let request = request_message(Operation::SendMessages, TRANSPORT, SESSION, 1, &[])
- .transmute_header(|header, new_header: &mut RequestHeader| {
+ .transmute_header(|header, new_header: &mut RoutedRequestHeader| {
*new_header = header;
- new_header.namespace = namespace.inner();
+ new_header.group = namespace.inner();
});
shard.on_message(request.into_generic()).await;
@@ -3467,9 +3478,9 @@ mod tests {
let namespace = IggyNamespace::new(0, 0, 0);
let request = request_message(Operation::SendMessages, TRANSPORT, SESSION, 1, &[])
- .transmute_header(|header, new_header: &mut RequestHeader| {
+ .transmute_header(|header, new_header: &mut RoutedRequestHeader| {
*new_header = header;
- new_header.namespace = namespace.inner();
+ new_header.group = namespace.inner();
});
// Namespace neither materialised nor tombstoned: the frame parks.
shard.on_message(request.into_generic()).await;
diff --git a/core/server-ng/src/dispatch/authz.rs b/core/server-ng/src/dispatch/authz.rs
index 9d48d9a46..47376ddcd 100644
--- a/core/server-ng/src/dispatch/authz.rs
+++ b/core/server-ng/src/dispatch/authz.rs
@@ -38,7 +38,9 @@ use iggy_binary_protocol::requests::consumer_groups::{
};
use iggy_binary_protocol::requests::streams::GetStreamRequest;
use iggy_binary_protocol::requests::topics::{GetTopicRequest, GetTopicsRequest};
-use iggy_binary_protocol::{Operation, PrepareHeader, RequestHeader, WireDecode, WireIdentifier};
+use iggy_binary_protocol::{
+ Operation, PrepareHeader, RoutedRequestHeader, WireDecode, WireIdentifier,
+};
use iggy_common::IggyError;
use journal::superblock::SuperblockStore;
use journal::{Journal, JournalHandle};
@@ -138,7 +140,7 @@ where
pub(super) async fn send_partition_deny_reply<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
transport_client_id: u128,
- request_header: &RequestHeader,
+ request_header: &RoutedRequestHeader,
status: u32,
) where
B: ShellBus,
@@ -298,7 +300,7 @@ where
#[allow(clippy::future_not_send)]
pub(super) async fn send_non_replicated_deny<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
- request: &Message<RequestHeader>,
+ request: &Message<RoutedRequestHeader>,
transport_client_id: u128,
status: u32,
) where
diff --git a/core/server-ng/src/http/session.rs b/core/server-ng/src/http/session.rs
index f6fd3df06..66f505cea 100644
--- a/core/server-ng/src/http/session.rs
+++ b/core/server-ng/src/http/session.rs
@@ -85,10 +85,10 @@ pub(in crate::http) struct HttpSession {
pub(in crate::http) key: String,
/// Shard-0 client id minted for this credential; its top 16 bits are 0, so
/// it shares the shard-0 id space with TCP virtual clients without
- /// colliding. Fills `RequestHeader.client` on every write.
+ /// colliding. Fills `RoutedRequestHeader.client` on every write.
pub(in crate::http) client_id: u128,
/// Cluster session number returned by the VSR `Register` commit. Fills
- /// `RequestHeader.session` on every write.
+ /// `RoutedRequestHeader.session` on every write.
pub(in crate::http) session: u64,
/// User the credential authenticated as. Consumed by the write path for
/// authorization.
diff --git a/core/server-ng/src/http/submit.rs b/core/server-ng/src/http/submit.rs
index e50b3373e..0843daa2e 100644
--- a/core/server-ng/src/http/submit.rs
+++ b/core/server-ng/src/http/submit.rs
@@ -26,7 +26,7 @@ use bytes::Bytes;
use consensus::MetadataHandle;
use futures::channel::oneshot;
use iggy_binary_protocol::consensus::Command2;
-use iggy_binary_protocol::{GenericHeader, Operation, RequestHeader};
+use iggy_binary_protocol::{GenericHeader, Operation, RoutedRequestHeader};
use iggy_common::IggyError;
use metadata::impls::metadata::StreamsFrontend;
use server_common::Message;
@@ -97,7 +97,7 @@ pub(in crate::http) async fn submit_committed(
session: &Rc<HttpSession>,
operation: Operation,
body: &[u8],
-) -> Result<(RequestHeader, Message<GenericHeader>, Option<String>), WriteError> {
+) -> Result<(RoutedRequestHeader, Message<GenericHeader>, Option<String>), WriteError> {
// Control writes are authorized in-apply on the metadata STM: a denial
// comes back as `Unauthorized` in the committed result section, which
// `committed_payload` maps to a 403. No pre-submit gate here, so the
@@ -182,7 +182,7 @@ async fn submit_gated(
operation: Operation,
max_tokens_per_user: u32,
body: &[u8],
-) -> Result<(RequestHeader, Message<GenericHeader>, Option<String>), WriteError> {
+) -> Result<(RoutedRequestHeader, Message<GenericHeader>, Option<String>), WriteError> {
let mut next_request_id = session.gate.lock().await;
// Burn the id at stamp time: every exit below (rewrite rejection,
// unresolved delete-segments, unanswered submit, exhausted transient
@@ -459,7 +459,7 @@ pub(in crate::http) async fn produce_unacked(
/// Install this session's in-process reply target on first data-plane use.
///
/// The registry key is the session's shard-0 client id - the same id stamped
-/// into `RequestHeader.client` - so a partition reply routed through
+/// into `RoutedRequestHeader.client` - so a partition reply routed through
/// `send_to_client` lands on this entry and resolves the request-keyed slot.
/// `None` from the registry means the key is already occupied; treat it as
/// installed but leave the token unset so this session never tears down an
diff --git a/core/server-ng/src/http/wire.rs b/core/server-ng/src/http/wire.rs
index e96866f80..c7026d09f 100644
--- a/core/server-ng/src/http/wire.rs
+++ b/core/server-ng/src/http/wire.rs
@@ -16,7 +16,7 @@
// under the License.
//! HTTP -> wire request mappers: produce/poll/consumer-offset encoders and
-//! the control-plane [`Message<RequestHeader>`] builder shared by the write path.
+//! the control-plane [`Message<RoutedRequestHeader>`] builder shared by the write path.
use bytes::{Bytes, BytesMut};
use iggy_binary_protocol::consensus::{Command2, HEADER_SIZE};
@@ -28,7 +28,7 @@ use iggy_binary_protocol::requests::consumer_offsets::{
use iggy_binary_protocol::requests::messages::{
PollMessagesRequest, RawMessage, SendMessagesEncoder,
};
-use iggy_binary_protocol::{AckLevel, Operation, RequestHeader};
+use iggy_binary_protocol::{AckLevel, Operation, RoutedRequestHeader};
use iggy_common::get_consumer_offset::GetConsumerOffset;
use iggy_common::poll_messages::DEFAULT_PARTITION_ID;
use iggy_common::store_consumer_offset::StoreConsumerOffset;
@@ -179,7 +179,7 @@ pub(in crate::http) const fn resync_required_polled_messages() -> PolledMessages
}
}
-/// Build a `Message<RequestHeader>` for a control-plane write by filling a zeroed
+/// Build a `Message<RoutedRequestHeader>` for a control-plane write by filling a zeroed
/// `#[repr(C)]` header, mirroring `wire::rewrite_request_body` and the partition
/// reconciler's prepare builder. `body` is the already-encoded wire request,
/// copied in after the header.
@@ -189,14 +189,14 @@ pub(in crate::http) fn build_request_message(
session_id: u64,
request_id: u64,
body: &[u8],
-) -> Message<RequestHeader> {
+) -> Message<RoutedRequestHeader> {
let total = HEADER_SIZE + body.len();
- let mut message = Message::<RequestHeader>::new(total);
+ let mut message = Message::<RoutedRequestHeader>::new(total);
message.as_mut_slice()[HEADER_SIZE..].copy_from_slice(body);
- let header = bytemuck::checked::try_from_bytes_mut::<RequestHeader>(
+ let header = bytemuck::checked::try_from_bytes_mut::<RoutedRequestHeader>(
&mut message.as_mut_slice()[..HEADER_SIZE],
)
- .expect("zeroed bytes form a valid RequestHeader");
+ .expect("zeroed bytes form a valid RoutedRequestHeader");
header.command = Command2::Request;
header.operation = operation;
header.client = client_id;
diff --git a/core/server-ng/src/partition_reconciler.rs b/core/server-ng/src/partition_reconciler.rs
index 819465b9a..7130c156b 100644
--- a/core/server-ng/src/partition_reconciler.rs
+++ b/core/server-ng/src/partition_reconciler.rs
@@ -1133,7 +1133,7 @@ mod tests {
CreateTopicRequest, CreateTopicWithAssignmentsRequest, DeleteTopicRequest,
};
use iggy_binary_protocol::{
- Command2, GenericHeader, Operation, PrepareHeader, ReplyHeader, RequestHeader,
+ Command2, GenericHeader, Operation, PrepareHeader, ReplyHeader, RoutedRequestHeader,
WireIdentifier,
};
use message_bus::IggyMessageBus;
@@ -1219,7 +1219,7 @@ mod tests {
header.command = Command2::Prepare;
header.size = u32::try_from(header_size).expect("prepare size fits u32");
header.operation = Operation::SendMessages;
- header.namespace = namespace.inner();
+ header.group = namespace.inner();
header.op = op;
msg.into_generic()
}
@@ -1251,17 +1251,17 @@ mod tests {
namespace: IggyNamespace,
body_len: usize,
) -> Message<GenericHeader> {
- let header_size = size_of::<RequestHeader>();
+ let header_size = size_of::<RoutedRequestHeader>();
let total_size = header_size + body_len;
- let mut msg = Message::<RequestHeader>::new(total_size);
- let header = bytemuck::checked::try_from_bytes_mut::<RequestHeader>(
+ let mut msg = Message::<RoutedRequestHeader>::new(total_size);
+ let header = bytemuck::checked::try_from_bytes_mut::<RoutedRequestHeader>(
&mut msg.as_mut_slice()[..header_size],
)
- .expect("zeroed bytes form a valid RequestHeader");
+ .expect("zeroed bytes form a valid RoutedRequestHeader");
header.command = Command2::Request;
header.size = u32::try_from(total_size).expect("request size fits u32");
header.operation = Operation::SendMessages;
- header.namespace = namespace.inner();
+ header.group = namespace.inner();
// Header validation rejects a zero session / request on a non-register
// op, and the park path runs after that validation.
header.session = 1;
diff --git a/core/server-ng/src/pat.rs b/core/server-ng/src/pat.rs
index dc22f2000..26c8ea6db 100644
--- a/core/server-ng/src/pat.rs
+++ b/core/server-ng/src/pat.rs
@@ -27,7 +27,7 @@ use iggy_binary_protocol::requests::personal_access_tokens::{
CreatePersonalAccessTokenRequest as WireCreatePersonalAccessTokenRequest,
DeletePersonalAccessTokenRequest as WireDeletePersonalAccessTokenRequest,
};
-use iggy_binary_protocol::{Operation, RequestHeader, WireDecode, WireEncode};
+use iggy_binary_protocol::{Operation, RoutedRequestHeader, WireDecode, WireEncode};
use iggy_common::IggyError;
use metadata::stm::user::{
CreatePersonalAccessTokenRequest as ReplicatedCreatePersonalAccessTokenRequest,
@@ -42,8 +42,8 @@ pub(crate) fn maybe_rewrite_pat_request(
transport_client_id: u128,
max_tokens_per_user: u32,
pat_count_of: impl FnOnce(u32) -> usize,
- request: Message<RequestHeader>,
-) -> Result<(Message<RequestHeader>, Option<String>), IggyError> {
+ request: Message<RoutedRequestHeader>,
+) -> Result<(Message<RoutedRequestHeader>, Option<String>), IggyError> {
let user_id = match request.header().operation {
Operation::CreatePersonalAccessToken | Operation::DeletePersonalAccessToken => sessions
.borrow()
@@ -69,8 +69,8 @@ pub(crate) fn rewrite_pat_request_for_user(
user_id: u32,
max_tokens_per_user: u32,
pat_count_of: impl FnOnce(u32) -> usize,
- request: Message<RequestHeader>,
-) -> Result<(Message<RequestHeader>, Option<String>), IggyError> {
+ request: Message<RoutedRequestHeader>,
+) -> Result<(Message<RoutedRequestHeader>, Option<String>), IggyError> {
let body = request_body(&request);
let mut raw_token = None;
let rewritten = match request.header().operation {
@@ -151,10 +151,10 @@ mod tests {
const MAX_TOKENS: u32 = 3;
const AT_LIMIT_TOKENS: [(&str, u8); 3] = [("one", b'a'), ("two", b'b'), ("three", b'c')];
- fn create_pat_request(name: &str) -> Message<RequestHeader> {
- let header_len = std::mem::size_of::<RequestHeader>();
- let mut template = Message::<RequestHeader>::new(header_len);
- let header = bytemuck::checked::try_from_bytes_mut::<RequestHeader>(
+ fn create_pat_request(name: &str) -> Message<RoutedRequestHeader> {
+ let header_len = std::mem::size_of::<RoutedRequestHeader>();
+ let mut template = Message::<RoutedRequestHeader>::new(header_len);
+ let header = bytemuck::checked::try_from_bytes_mut::<RoutedRequestHeader>(
&mut template.as_mut_slice()[..header_len],
)
.expect("zeroed bytes are a valid request header");
@@ -266,9 +266,9 @@ mod tests {
#[test]
fn given_delete_op_when_at_limit_should_pass_the_gate() {
- let header_len = std::mem::size_of::<RequestHeader>();
- let mut template = Message::<RequestHeader>::new(header_len);
- let header = bytemuck::checked::try_from_bytes_mut::<RequestHeader>(
+ let header_len = std::mem::size_of::<RoutedRequestHeader>();
+ let mut template = Message::<RoutedRequestHeader>::new(header_len);
+ let header = bytemuck::checked::try_from_bytes_mut::<RoutedRequestHeader>(
&mut template.as_mut_slice()[..header_len],
)
.expect("zeroed bytes are a valid request header");
diff --git a/core/server-ng/src/responses.rs b/core/server-ng/src/responses.rs
index c661a375c..900148638 100644
--- a/core/server-ng/src/responses.rs
+++ b/core/server-ng/src/responses.rs
@@ -76,7 +76,7 @@ use iggy_binary_protocol::responses::users::get_users::GetUsersResponse;
use iggy_binary_protocol::responses::users::user_response::UserResponse;
use iggy_binary_protocol::{
Command2, GenericHeader, IGGY_PROTOCOL_VERSION, KIND_CONSUMER_GROUP, Operation, ReplyHeader,
- RequestHeader, WireDecode, WireEncode, WireIdentifier, WireName, WirePartitioning,
+ RoutedRequestHeader, WireDecode, WireEncode, WireIdentifier, WireName, WirePartitioning,
};
use iggy_common::{EncryptorKind, Identifier, IggyError, IggyExpiry, IggyTimestamp, MaxTopicSize};
use journal::superblock::SuperblockStore;
@@ -1236,7 +1236,7 @@ pub(crate) enum NonReplicatedResponse {
impl NonReplicatedResponse {
pub(crate) fn into_reply(
self,
- request_header: &RequestHeader,
+ request_header: &RoutedRequestHeader,
client_id: u128,
session: u64,
commit: u64,
@@ -1251,7 +1251,7 @@ impl NonReplicatedResponse {
}
pub(crate) fn build_empty_reply(
- request_header: &RequestHeader,
+ request_header: &RoutedRequestHeader,
client_id: u128,
session: u64,
commit: u64,
@@ -1267,7 +1267,7 @@ pub(crate) fn build_empty_reply(
/// reply, and only the partition primary's pre-pipeline deny pins it to 0,
/// stamped through `consensus::build_deny_reply_from_request`.
pub(crate) fn build_deny_reply(
- request_header: &RequestHeader,
+ request_header: &RoutedRequestHeader,
client_id: u128,
session: u64,
commit: u64,
@@ -1287,7 +1287,7 @@ pub(crate) fn build_deny_reply(
const SERVER_VERSION: &str = env!("CARGO_PKG_VERSION");
pub(crate) fn build_login_register_reply(
- request_header: &RequestHeader,
+ request_header: &RoutedRequestHeader,
client_id: u128,
session: u64,
commit: u64,
@@ -1318,7 +1318,7 @@ pub(crate) fn build_login_register_reply(
}
pub(crate) fn build_reply_from_bytes(
- request_header: &RequestHeader,
+ request_header: &RoutedRequestHeader,
client_id: u128,
session: u64,
commit: u64,
@@ -1341,7 +1341,7 @@ pub(crate) fn build_reply_from_bytes(
/// (no token, a committed business rejection, or an eviction frame) the
/// committed reply passes through unchanged.
pub(crate) fn build_raw_pat_reply(
- request_header: &RequestHeader,
+ request_header: &RoutedRequestHeader,
committed: Message<GenericHeader>,
raw_token: Option<String>,
) -> Result<Message<GenericHeader>, IggyError> {
@@ -1402,7 +1402,7 @@ pub(crate) fn build_raw_pat_reply(
}
pub(crate) fn build_reply_with_body(
- request_header: &RequestHeader,
+ request_header: &RoutedRequestHeader,
client_id: u128,
session: u64,
commit: u64,
@@ -1431,7 +1431,6 @@ pub(crate) fn build_reply_with_body(
timestamp: request_header.timestamp,
request: request_header.request,
operation: request_header.operation,
- namespace: request_header.namespace,
..Default::default()
};
write_body(&mut reply.as_mut_slice()[header_len..total_size]);
@@ -1623,10 +1622,10 @@ pub(crate) fn build_consumer_offset_body(
mod tests {
use super::*;
- fn pat_request_header() -> RequestHeader {
- let zeroed = [0u8; std::mem::size_of::<RequestHeader>()];
- let mut header = *bytemuck::checked::try_from_bytes::<RequestHeader>(&zeroed)
- .expect("zeroed bytes form a valid RequestHeader");
+ fn pat_request_header() -> RoutedRequestHeader {
+ let zeroed = [0u8; std::mem::size_of::<RoutedRequestHeader>()];
+ let mut header = *bytemuck::checked::try_from_bytes::<RoutedRequestHeader>(&zeroed)
+ .expect("zeroed bytes form a valid RoutedRequestHeader");
header.command = Command2::Request;
header.operation = Operation::CreatePersonalAccessToken;
header.client = 42;
diff --git a/core/server-ng/src/session_manager.rs b/core/server-ng/src/session_manager.rs
index 9033aa98b..18338c42a 100644
--- a/core/server-ng/src/session_manager.rs
+++ b/core/server-ng/src/session_manager.rs
@@ -50,8 +50,8 @@ pub enum ConnectionState {
Authenticated { user_id: u32 },
/// Register committed through consensus. Connection is bound to a
/// `(client_id, session)` pair. Requests on this connection use
- /// these values to populate `RequestHeader.client` and
- /// `RequestHeader.session`.
+ /// these values to populate `RoutedRequestHeader.client` and
+ /// `RoutedRequestHeader.session`.
Bound {
user_id: u32,
client_id: u128,
diff --git a/core/server-ng/src/users.rs b/core/server-ng/src/users.rs
index 94a5cb270..802d092e3 100644
--- a/core/server-ng/src/users.rs
+++ b/core/server-ng/src/users.rs
@@ -41,7 +41,7 @@ use bytes::Bytes;
use consensus::MetadataHandle;
use iggy_binary_protocol::codec::{WireDecode, WireEncode};
use iggy_binary_protocol::requests::users::{ChangePasswordRequest, CreateUserRequest};
-use iggy_binary_protocol::{Operation, PrepareHeader, RequestHeader};
+use iggy_binary_protocol::{Operation, PrepareHeader, RoutedRequestHeader};
use iggy_common::IggyError;
use journal::superblock::SuperblockStore;
use journal::{Journal, JournalHandle};
@@ -60,8 +60,8 @@ use std::rc::Rc;
/// undecodable password body.
pub(crate) fn maybe_rewrite_user_password_request<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
- request: Message<RequestHeader>,
-) -> Result<Message<RequestHeader>, IggyError>
+ request: Message<RoutedRequestHeader>,
+) -> Result<Message<RoutedRequestHeader>, IggyError>
where
B: ShellBus,
MJ: JournalHandle + 'static,
diff --git a/core/server-ng/src/wire.rs b/core/server-ng/src/wire.rs
index e7a047089..c4b37a6f1 100644
--- a/core/server-ng/src/wire.rs
+++ b/core/server-ng/src/wire.rs
@@ -21,13 +21,13 @@
//! transport-kind discriminant mapping.
use bytes::Bytes;
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use iggy_common::IggyError;
use message_bus::installer::conn_info::ClientTransportKind;
use server_common::Message;
-pub(crate) fn request_body(request: &Message<RequestHeader>) -> &[u8] {
- &request.as_slice()[std::mem::size_of::<RequestHeader>()..request.header().size as usize]
+pub(crate) fn request_body(request: &Message<RoutedRequestHeader>) -> &[u8] {
+ &request.as_slice()[std::mem::size_of::<RoutedRequestHeader>()..request.header().size as usize]
}
/// Map the transport kind to the legacy wire discriminant
@@ -51,21 +51,21 @@ pub(crate) fn usize_to_u32(value: usize) -> Result<u32, IggyError> {
/// request rewrites that swap a secret-bearing wire body for the
/// hash-carrying replicated body before consensus.
pub(crate) fn rewrite_request_body(
- request: &Message<RequestHeader>,
+ request: &Message<RoutedRequestHeader>,
body: &Bytes,
-) -> Result<Message<RequestHeader>, IggyError> {
- let total_size = std::mem::size_of::<RequestHeader>()
+) -> Result<Message<RoutedRequestHeader>, IggyError> {
+ let total_size = std::mem::size_of::<RoutedRequestHeader>()
.checked_add(body.len())
.ok_or(IggyError::InvalidConfiguration)?;
let size = u32::try_from(total_size).map_err(|_| IggyError::InvalidConfiguration)?;
- let mut rewritten = Message::<RequestHeader>::new(total_size);
- let header = bytemuck::checked::try_from_bytes_mut::<RequestHeader>(
- &mut rewritten.as_mut_slice()[..std::mem::size_of::<RequestHeader>()],
+ let mut rewritten = Message::<RoutedRequestHeader>::new(total_size);
+ let header = bytemuck::checked::try_from_bytes_mut::<RoutedRequestHeader>(
+ &mut rewritten.as_mut_slice()[..std::mem::size_of::<RoutedRequestHeader>()],
)
.expect("zeroed bytes are a valid request header");
*header = *request.header();
header.size = size;
- rewritten.as_mut_slice()[std::mem::size_of::<RequestHeader>()..].copy_from_slice(body);
+ rewritten.as_mut_slice()[std::mem::size_of::<RoutedRequestHeader>()..].copy_from_slice(body);
// TODO(vsr): the body changed but `request_checksum` / `checksum` /
// `checksum_body` were copied verbatim from the original header. Safe
// today because the SDK initializes `request_checksum` to 0 and the
diff --git a/core/server_common/src/consensus_message.rs b/core/server_common/src/consensus_message.rs
index 4fe65638b..7b3cf8a57 100644
--- a/core/server_common/src/consensus_message.rs
+++ b/core/server_common/src/consensus_message.rs
@@ -19,9 +19,9 @@ use crate::iobuf::{Frozen, Owned};
use iggy_binary_protocol::{
Command2, CommitHeader, ConsensusError, ConsensusHeader, DoViewChangeHeader, GenericHeader,
Operation, PrepareHeader, PrepareOkHeader, RepairPrepareHeader, RepairRangeReplyHeader,
- RequestHeader, RequestPreparesHeader, RequestStartViewHeader, RequestStateChunkHeader,
- RequestStateTransferHeader, StartViewChangeHeader, StartViewHeader, StateChunkHeader,
- StateTransferTargetHeader,
+ RequestPreparesHeader, RequestStartViewHeader, RequestStateChunkHeader,
+ RequestStateTransferHeader, RoutedRequestHeader, StartViewChangeHeader, StartViewHeader,
+ StateChunkHeader, StateTransferTargetHeader,
};
use smallvec::SmallVec;
use std::{marker::PhantomData, mem::size_of};
@@ -495,7 +495,7 @@ where
#[derive(Debug)]
pub enum MessageBag {
- Request(Message<RequestHeader>),
+ Request(Message<RoutedRequestHeader>),
Prepare(Message<PrepareHeader>),
PrepareOk(Message<PrepareOkHeader>),
StartViewChange(Message<StartViewChangeHeader>),
@@ -600,7 +600,9 @@ where
match command {
Command2::Prepare => Ok(Self::Prepare(value.try_into_typed::<PrepareHeader>()?)),
- Command2::Request => Ok(Self::Request(value.try_into_typed::<RequestHeader>()?)),
+ Command2::Request => Ok(Self::Request(
+ value.try_into_typed::<RoutedRequestHeader>()?,
+ )),
Command2::PrepareOk => Ok(Self::PrepareOk(value.try_into_typed::<PrepareOkHeader>()?)),
Command2::StartViewChange => Ok(Self::StartViewChange(
value.try_into_typed::<StartViewChangeHeader>()?,
@@ -659,12 +661,12 @@ mod tests {
// Field offsets via `offset_of!`: a field reorder fails to compile here
// rather than silently corrupting test bytes.
- const SIZE_OFF: usize = std::mem::offset_of!(RequestHeader, size);
- const COMMAND_OFF: usize = std::mem::offset_of!(RequestHeader, command);
- const REQUEST_CLIENT_OFF: usize = std::mem::offset_of!(RequestHeader, client);
- const REQUEST_OPERATION_OFF: usize = std::mem::offset_of!(RequestHeader, operation);
- const REQUEST_SESSION_OFF: usize = std::mem::offset_of!(RequestHeader, session);
- const REQUEST_REQUEST_OFF: usize = std::mem::offset_of!(RequestHeader, request);
+ const SIZE_OFF: usize = std::mem::offset_of!(RoutedRequestHeader, size);
+ const COMMAND_OFF: usize = std::mem::offset_of!(RoutedRequestHeader, command);
+ const REQUEST_CLIENT_OFF: usize = std::mem::offset_of!(RoutedRequestHeader, client);
+ const REQUEST_OPERATION_OFF: usize = std::mem::offset_of!(RoutedRequestHeader, operation);
+ const REQUEST_SESSION_OFF: usize = std::mem::offset_of!(RoutedRequestHeader, session);
+ const REQUEST_REQUEST_OFF: usize = std::mem::offset_of!(RoutedRequestHeader, request);
fn header_bytes(command: Command2, size: u32) -> Owned<MESSAGE_ALIGN> {
header_bytes_sized(command, size, 256)
@@ -737,7 +739,7 @@ mod tests {
#[test]
#[should_panic(expected = "size must be at least header size")]
fn message_new_smaller_than_header_panics() {
- let _ = Message::<RequestHeader>::new(100);
+ let _ = Message::<RoutedRequestHeader>::new(100);
}
// try_from(Owned): validation gates the unsafe construction
@@ -745,7 +747,7 @@ mod tests {
#[test]
fn try_from_owned_too_short_returns_err() {
let owned = Owned::<MESSAGE_ALIGN>::zeroed(100);
- let result = Message::<RequestHeader>::try_from(owned);
+ let result = Message::<RoutedRequestHeader>::try_from(owned);
assert!(matches!(result, Err(ConsensusError::InvalidCommand { .. })));
}
@@ -753,13 +755,13 @@ mod tests {
fn try_from_owned_invalid_bit_pattern_returns_err() {
let mut owned = Owned::<MESSAGE_ALIGN>::zeroed(256);
owned.as_mut_slice()[COMMAND_OFF] = 99; // outside Command2's discriminant range
- let result = Message::<RequestHeader>::try_from(owned);
+ let result = Message::<RoutedRequestHeader>::try_from(owned);
assert!(matches!(result, Err(ConsensusError::InvalidBitPattern)));
}
#[test]
fn try_from_owned_buffer_shorter_than_claimed_size_returns_err() {
- // Header parses cleanly (RequestHeader::validate doesn't gate on
+ // Header parses cleanly (RoutedRequestHeader::validate doesn't gate on
// size), but the encoded `size` field claims more bytes than the
// backing buffer holds. The buffer-bounds check at the bottom of
// `Message::try_from` must reject. (Both this case and the
@@ -769,7 +771,7 @@ mod tests {
let owned = header_bytes(Command2::Request, 999);
// header_bytes already produces a 256-byte buffer; size=999 > 256,
// so try_from rejects via `bytes.len() < header.size()`.
- let result = Message::<RequestHeader>::try_from(owned);
+ let result = Message::<RoutedRequestHeader>::try_from(owned);
assert!(matches!(result, Err(ConsensusError::InvalidCommand { .. })));
}
@@ -779,8 +781,11 @@ mod tests {
// so only the construction-time `size` floor rejects it (the
// buffer-length check passes). Guards the `[size_of::<H>()..size]`
// underflow at every downstream call site.
- let owned = header_bytes(Command2::Request, size_of::<RequestHeader>() as u32 - 1);
- let result = Message::<RequestHeader>::try_from(owned);
+ let owned = header_bytes(
+ Command2::Request,
+ size_of::<RoutedRequestHeader>() as u32 - 1,
+ );
+ let result = Message::<RoutedRequestHeader>::try_from(owned);
assert!(matches!(result, Err(ConsensusError::InvalidCommand { .. })));
}
@@ -789,7 +794,7 @@ mod tests {
#[test]
fn as_generic_view_reads_command_byte() {
let owned = header_bytes(Command2::Request, 256);
- let typed = Message::<RequestHeader>::try_from(owned).expect("valid");
+ let typed = Message::<RoutedRequestHeader>::try_from(owned).expect("valid");
let generic = typed.as_generic();
assert_eq!(generic.header().command, Command2::Request);
assert_eq!(generic.total_len(), 256);
@@ -799,11 +804,11 @@ mod tests {
#[test]
fn try_as_typed_command_mismatch_returns_err_without_unsafe_cast() {
- // bytes are a valid Prepare; asking for RequestHeader must fail
+ // bytes are a valid Prepare; asking for RoutedRequestHeader must fail
// *before* the unsafe ptr-cast inside try_as_typed.
let owned = header_bytes(Command2::Prepare, 256);
let generic = Message::<GenericHeader>::try_from(owned).expect("valid");
- let result = generic.try_as_typed::<RequestHeader>();
+ let result = generic.try_as_typed::<RoutedRequestHeader>();
assert!(matches!(
result,
Err(ConsensusError::InvalidCommand {
@@ -815,7 +820,7 @@ mod tests {
#[test]
fn try_as_typed_invalid_validation_returns_err() {
- // RequestHeader::validate rejects operation=Register with non-zero session.
+ // RoutedRequestHeader::validate rejects operation=Register with non-zero session.
let mut owned = header_bytes(Command2::Request, 256);
{
let buf = owned.as_mut_slice();
@@ -823,7 +828,7 @@ mod tests {
buf[REQUEST_SESSION_OFF..REQUEST_SESSION_OFF + 8].copy_from_slice(&5u64.to_le_bytes());
}
let generic = Message::<GenericHeader>::try_from(owned).expect("valid generic");
- let result = generic.try_as_typed::<RequestHeader>();
+ let result = generic.try_as_typed::<RoutedRequestHeader>();
assert!(matches!(result, Err(ConsensusError::InvalidField(_))));
}
@@ -833,7 +838,7 @@ mod tests {
fn try_into_typed_command_mismatch_returns_err() {
let owned = header_bytes(Command2::Prepare, 256);
let generic = Message::<GenericHeader>::try_from(owned).expect("valid");
- let result = generic.try_into_typed::<RequestHeader>();
+ let result = generic.try_into_typed::<RoutedRequestHeader>();
assert!(matches!(
result,
Err(ConsensusError::InvalidCommand {
@@ -898,7 +903,7 @@ mod tests {
#[test]
fn messagebag_dispatch_request_with_invalid_register_session_returns_err() {
- // `RequestHeader::validate` rejects Register with non-zero session.
+ // `RoutedRequestHeader::validate` rejects Register with non-zero session.
let mut owned = header_bytes(Command2::Request, 256);
{
let buf = owned.as_mut_slice();
@@ -912,7 +917,7 @@ mod tests {
// Ingress validation runs on every client frame at the network boundary,
// reached through `MessageBag::try_from` -> `try_into_typed` ->
- // `RequestHeader::validate`. Several dedup and authz conclusions rest on
+ // `RoutedRequestHeader::validate`. Several dedup and authz conclusions rest on
// it running, so pin the field rules rather than the plumbing: whatever
// `request_preflight` and the operation gate see downstream has already
// passed these.
@@ -978,7 +983,7 @@ mod tests {
#[test]
fn request_message_deep_copy_independent() {
let owned = header_bytes(Command2::Request, 256);
- let mut msg = Message::<RequestHeader>::try_from(owned).expect("valid");
+ let mut msg = Message::<RoutedRequestHeader>::try_from(owned).expect("valid");
let copy = msg.deep_copy();
// Mutate the original's bytes; the deep copy must be untouched.
msg.as_mut_slice()[200] = 0xab;
@@ -991,7 +996,7 @@ mod tests {
#[test]
fn transmute_header_request_to_prepare() {
let owned = header_bytes(Command2::Request, 256);
- let msg = Message::<RequestHeader>::try_from(owned).expect("valid");
+ let msg = Message::<RoutedRequestHeader>::try_from(owned).expect("valid");
let prepared: Message<PrepareHeader> =
msg.transmute_header::<PrepareHeader>(|_old, new| {
new.command = Command2::Prepare;
diff --git a/core/server_common/src/send_messages2.rs b/core/server_common/src/send_messages2.rs
index f0d9a7f88..6a67109f8 100644
--- a/core/server_common/src/send_messages2.rs
+++ b/core/server_common/src/send_messages2.rs
@@ -19,7 +19,7 @@ use crate::consensus_message::{MESSAGE_ALIGN, Message};
use crate::iobuf::Owned;
use crate::sharding::IggyNamespace;
use bytes::{Bytes, BytesMut};
-use iggy_binary_protocol::{PrepareHeader, RequestHeader};
+use iggy_binary_protocol::{PrepareHeader, RoutedRequestHeader};
use iggy_common::{EncryptorKind, INDEX_SIZE, IggyError, random_id};
use std::hash::Hasher;
use twox_hash::XxHash3_64;
@@ -196,20 +196,20 @@ impl SendMessages2Owned {
pub fn encode_request(
self,
- mut request_header: RequestHeader,
- ) -> Result<Message<RequestHeader>, IggyError> {
- let total_size = std::mem::size_of::<RequestHeader>() + self.header.total_size();
+ mut request_header: RoutedRequestHeader,
+ ) -> Result<Message<RoutedRequestHeader>, IggyError> {
+ let total_size = std::mem::size_of::<RoutedRequestHeader>() + self.header.total_size();
// The converted body differs in size from the legacy wire body the
// header described; a stale `size` truncates the rebuilt blob for
// every downstream slice (stamping, journal reads).
request_header.size = u32::try_from(total_size).map_err(|_| IggyError::InvalidCommand)?;
let mut buffer = Owned::<MESSAGE_ALIGN>::zeroed(total_size);
let bytes = buffer.as_mut_slice();
- bytes[0..std::mem::size_of::<RequestHeader>()]
+ bytes[0..std::mem::size_of::<RoutedRequestHeader>()]
.copy_from_slice(bytemuck::bytes_of(&request_header));
self.header.encode_into(
- &mut bytes[std::mem::size_of::<RequestHeader>()
- ..std::mem::size_of::<RequestHeader>() + COMMAND_HEADER_SIZE],
+ &mut bytes[std::mem::size_of::<RoutedRequestHeader>()
+ ..std::mem::size_of::<RoutedRequestHeader>() + COMMAND_HEADER_SIZE],
);
bytes[PREPARE_SPLIT_POINT..PREPARE_SPLIT_POINT + self.blob.len()]
.copy_from_slice(&self.blob);
@@ -473,12 +473,12 @@ pub(crate) type FrozenBatchHeader = crate::iobuf::Frozen<MESSAGE_ALIGN>;
/// [`IggyError::InvalidCommand`] on an undecodable batch; encryption errors
/// propagate from the encryptor.
pub fn encrypt_batch_request(
- message: Message<RequestHeader>,
+ message: Message<RoutedRequestHeader>,
encryptor: &EncryptorKind,
-) -> Result<Message<RequestHeader>, IggyError> {
+) -> Result<Message<RoutedRequestHeader>, IggyError> {
let request_header = *message.header();
let total_size = request_header.size as usize;
- let body = &message.as_slice()[std::mem::size_of::<RequestHeader>()..total_size];
+ let body = &message.as_slice()[std::mem::size_of::<RoutedRequestHeader>()..total_size];
let batch = decode_batch_slice(body)?;
let mut blob = BytesMut::with_capacity(batch.blob().len() * 2);
@@ -537,12 +537,12 @@ pub enum ChecksumMode {
pub fn convert_request_message(
namespace: IggyNamespace,
- message: Message<RequestHeader>,
+ message: Message<RoutedRequestHeader>,
checksum: ChecksumMode,
-) -> Result<Message<RequestHeader>, IggyError> {
+) -> Result<Message<RoutedRequestHeader>, IggyError> {
let request_header = *message.header();
let total_size = request_header.size as usize;
- let body = &message.as_slice()[std::mem::size_of::<RequestHeader>()..total_size];
+ let body = &message.as_slice()[std::mem::size_of::<RoutedRequestHeader>()..total_size];
// A canonical body enters the pipeline verbatim, so it must end exactly at
// `batch_length`: `size` and `batch_length` are independent client-supplied
// fields and `decode_batch_slice` only lower-bounds the frame. A suffix past
@@ -556,7 +556,7 @@ pub fn convert_request_message(
}
/// Transcode a legacy `SendMessages` request body directly into the canonical
-/// `[RequestHeader][256B SendMessages2Header][blob]` form, writing each message
+/// `[RoutedRequestHeader][256B SendMessages2Header][blob]` form, writing each message
/// record straight into the final aligned buffer.
///
/// Fused replacement for the `from_legacy_request(..).encode_request(..)`
@@ -571,9 +571,9 @@ pub fn convert_request_message(
fn transcode_legacy_request(
namespace: IggyNamespace,
body: &[u8],
- mut request_header: RequestHeader,
+ mut request_header: RoutedRequestHeader,
checksum: ChecksumMode,
-) -> Result<Message<RequestHeader>, IggyError> {
+) -> Result<Message<RoutedRequestHeader>, IggyError> {
let (message_count, messages) = legacy_messages_slice(body)?;
let mut parsed = Vec::with_capacity(message_count as usize);
let mut origin_timestamp = u64::MAX;
@@ -598,7 +598,7 @@ fn transcode_legacy_request(
origin_timestamp = 0;
}
- let header_size = std::mem::size_of::<RequestHeader>();
+ let header_size = std::mem::size_of::<RoutedRequestHeader>();
let batch_length = COMMAND_HEADER_SIZE
.checked_add(blob_len)
.ok_or(IggyError::InvalidCommand)?;
@@ -1381,14 +1381,14 @@ mod tests {
body
}
- fn legacy_request_message(body: &[u8]) -> Message<RequestHeader> {
- let header_size = std::mem::size_of::<RequestHeader>();
+ fn legacy_request_message(body: &[u8]) -> Message<RoutedRequestHeader> {
+ let header_size = std::mem::size_of::<RoutedRequestHeader>();
let total = header_size + body.len();
let mut buffer = Owned::<MESSAGE_ALIGN>::zeroed(total);
{
- let header: &mut RequestHeader =
+ let header: &mut RoutedRequestHeader =
bytemuck::checked::try_from_bytes_mut(&mut buffer.as_mut_slice()[..header_size])
- .expect("zeroed bytes form a valid RequestHeader");
+ .expect("zeroed bytes form a valid RoutedRequestHeader");
header.command = Command2::Request;
header.operation = Operation::SendMessages;
header.client = 1;
@@ -1420,7 +1420,7 @@ mod tests {
let legacy = legacy_request_message(&legacy_send_messages_body(&messages));
let converted = convert_request_message(namespace, legacy, ChecksumMode::Compute)
.expect("legacy body transcodes");
- let header_size = std::mem::size_of::<RequestHeader>();
+ let header_size = std::mem::size_of::<RoutedRequestHeader>();
let actual_body = &converted.as_slice()[header_size..converted.header().size as usize];
assert_eq!(
@@ -1447,7 +1447,7 @@ mod tests {
let namespace = IggyNamespace::new(1, 1, 3);
let messages = sample_messages();
let body = legacy_send_messages_body(&messages);
- let header_size = std::mem::size_of::<RequestHeader>();
+ let header_size = std::mem::size_of::<RoutedRequestHeader>();
let computed = convert_request_message(
namespace,
@@ -1490,7 +1490,7 @@ mod tests {
// batch and returns it unchanged. Every decode must succeed.
let namespace = IggyNamespace::new(1, 1, 3);
let messages = sample_messages();
- let header_size = std::mem::size_of::<RequestHeader>();
+ let header_size = std::mem::size_of::<RoutedRequestHeader>();
let legacy = legacy_request_message(&legacy_send_messages_body(&messages));
let canonical = convert_request_message(namespace, legacy, ChecksumMode::Compute)
@@ -1523,19 +1523,19 @@ mod tests {
const TRAILING_JUNK_CASES: [&[u8]; 2] = [&[0xAA], &[0xFF; 64]];
/// Canonical `SendMessages` request carrying `junk` past `batch_length`, with
- /// `RequestHeader.size` inflated to cover it. `size` and `batch_length` are
+ /// `RoutedRequestHeader.size` inflated to cover it. `size` and `batch_length` are
/// independent wire fields, so a non-conforming client can emit this.
- fn canonical_request_with_trailing_bytes(junk: &[u8]) -> Message<RequestHeader> {
+ fn canonical_request_with_trailing_bytes(junk: &[u8]) -> Message<RoutedRequestHeader> {
let namespace = IggyNamespace::new(1, 1, 3);
let owned =
SendMessages2Owned::from_messages(namespace, &sample_messages()).expect("build batch");
- let header_size = std::mem::size_of::<RequestHeader>();
+ let header_size = std::mem::size_of::<RoutedRequestHeader>();
let total = header_size + owned.header.total_size() + junk.len();
let mut buffer = Owned::<MESSAGE_ALIGN>::zeroed(total);
{
- let header: &mut RequestHeader =
+ let header: &mut RoutedRequestHeader =
bytemuck::checked::try_from_bytes_mut(&mut buffer.as_mut_slice()[..header_size])
- .expect("zeroed bytes form a valid RequestHeader");
+ .expect("zeroed bytes form a valid RoutedRequestHeader");
header.command = Command2::Request;
header.operation = Operation::SendMessages;
header.client = 1;
diff --git a/core/server_common/src/sharding/mod.rs b/core/server_common/src/sharding/mod.rs
index 761e41d04..b73169f52 100644
--- a/core/server_common/src/sharding/mod.rs
+++ b/core/server_common/src/sharding/mod.rs
@@ -22,10 +22,9 @@ mod shard_id;
pub use local_idx::LocalIdx;
pub use namespace::{
- IggyNamespace, MAX_PARTITIONS, MAX_STREAMS, MAX_TOPICS, METADATA_CONSENSUS_NAMESPACE,
- NamespaceCapacityError, PACKED_NAMESPACE_BITS, PACKED_NAMESPACE_MAX, PARTITION_BITS,
- PARTITION_MASK, PARTITION_SHIFT, STREAM_BITS, STREAM_MASK, STREAM_SHIFT, TOPIC_BITS,
- TOPIC_MASK, TOPIC_SHIFT,
+ IggyNamespace, MAX_PARTITIONS, MAX_STREAMS, MAX_TOPICS, METADATA_GROUP, NamespaceCapacityError,
+ PACKED_NAMESPACE_BITS, PACKED_NAMESPACE_MAX, PARTITION_BITS, PARTITION_MASK, PARTITION_SHIFT,
+ STREAM_BITS, STREAM_MASK, STREAM_SHIFT, TOPIC_BITS, TOPIC_MASK, TOPIC_SHIFT,
};
pub use partition_location::PartitionLocation;
pub use shard_id::ShardId;
diff --git a/core/server_common/src/sharding/namespace.rs b/core/server_common/src/sharding/namespace.rs
index 839edc0a2..435b076c4 100644
--- a/core/server_common/src/sharding/namespace.rs
+++ b/core/server_common/src/sharding/namespace.rs
@@ -27,7 +27,7 @@
// shard. Re-exported here for ergonomics of existing call sites.
pub use iggy_binary_protocol::namespace::{
- MAX_PARTITIONS, MAX_STREAMS, MAX_TOPICS, METADATA_CONSENSUS_NAMESPACE, PACKED_NAMESPACE_BITS,
+ MAX_PARTITIONS, MAX_STREAMS, MAX_TOPICS, METADATA_GROUP, PACKED_NAMESPACE_BITS,
PACKED_NAMESPACE_MAX, PARTITION_BITS, PARTITION_MASK, PARTITION_SHIFT, STREAM_BITS,
STREAM_MASK, STREAM_SHIFT, TOPIC_BITS, TOPIC_MASK, TOPIC_SHIFT, bits_required,
};
@@ -149,7 +149,7 @@ impl IggyNamespace {
#[cfg(test)]
mod tests {
use super::{
- IggyNamespace, MAX_PARTITIONS, MAX_STREAMS, MAX_TOPICS, METADATA_CONSENSUS_NAMESPACE,
+ IggyNamespace, MAX_PARTITIONS, MAX_STREAMS, MAX_TOPICS, METADATA_GROUP,
NamespaceCapacityError, PACKED_NAMESPACE_BITS, PACKED_NAMESPACE_MAX,
};
@@ -158,26 +158,26 @@ mod tests {
// STREAM_BITS / TOPIC_BITS / PARTITION_BITS that closes the gap between
// the packed range and the sentinel will fail to build.
const _: () = {
- assert!(METADATA_CONSENSUS_NAMESPACE > PACKED_NAMESPACE_MAX);
+ assert!(METADATA_GROUP > PACKED_NAMESPACE_MAX);
assert!(PACKED_NAMESPACE_BITS == 12 + 12 + 20);
assert!(PACKED_NAMESPACE_MAX == (1u64 << PACKED_NAMESPACE_BITS) - 1);
};
#[test]
fn metadata_sentinel_cannot_collide_with_any_packable_namespace() {
- assert!(!IggyNamespace::is_packable(METADATA_CONSENSUS_NAMESPACE));
+ assert!(!IggyNamespace::is_packable(METADATA_GROUP));
// The (0, 0, 0) corner is intentionally a legal partition, which is
// precisely why `0` is unsuitable as the metadata sentinel.
let zero = IggyNamespace::new(0, 0, 0);
assert_eq!(zero.inner(), 0);
assert!(IggyNamespace::is_packable(zero.inner()));
- assert_ne!(zero.inner(), METADATA_CONSENSUS_NAMESPACE);
+ assert_ne!(zero.inner(), METADATA_GROUP);
// Maximum packable triple stays inside the packed range.
let max = IggyNamespace::new(MAX_STREAMS - 1, MAX_TOPICS - 1, MAX_PARTITIONS - 1);
assert!(IggyNamespace::is_packable(max.inner()));
- assert_ne!(max.inner(), METADATA_CONSENSUS_NAMESPACE);
+ assert_ne!(max.inner(), METADATA_GROUP);
}
#[test]
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index b7d291b0c..1c3a49950 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -38,9 +38,9 @@ use crossfire::TrySendError;
use futures::FutureExt;
use iggy_binary_protocol::{
Command2, CommitHeader, DoViewChangeHeader, GenericHeader, Operation, PrepareHeader,
- PrepareOkHeader, RepairPrepareHeader, RepairRangeReplyHeader, RequestHeader,
- RequestPreparesHeader, RequestStartViewHeader, RequestStateChunkHeader,
- RequestStateTransferHeader, StartViewChangeHeader, StartViewHeader, StateChunkHeader,
+ PrepareOkHeader, RepairPrepareHeader, RepairRangeReplyHeader, RequestPreparesHeader,
+ RequestStartViewHeader, RequestStateChunkHeader, RequestStateTransferHeader,
+ RoutedRequestHeader, StartViewChangeHeader, StartViewHeader, StateChunkHeader,
StateTransferTargetHeader,
};
#[cfg(any(test, feature = "simulator"))]
@@ -2008,7 +2008,7 @@ where
{
match MessageBag::try_from(message) {
Ok(MessageBag::Request(request)) => {
- let routing = (request.header().operation, request.header().namespace);
+ let routing = (request.header().operation, request.header().group);
match self.park_if_unmaterialised(request, routing.0, routing.1) {
// The incarnation fence runs only here, on client traffic.
// A backup denying what the primary admitted would diverge
@@ -2033,7 +2033,7 @@ where
}
}
Ok(MessageBag::Prepare(prepare)) => {
- let routing = (prepare.header().operation, prepare.header().namespace);
+ let routing = (prepare.header().operation, prepare.header().group);
// 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.
@@ -2469,7 +2469,7 @@ where
/// `frame_drops_total{variant=partition,reason=park_dropped}`.
fn deny_parked_client_request(&self, frame: ParkedFrame) -> bool {
if frame.message.header().command == Command2::Request
- && let Ok(request) = frame.message.try_into_typed::<RequestHeader>()
+ && let Ok(request) = frame.message.try_into_typed::<RoutedRequestHeader>()
{
return self.stage_transient_deny(request.header());
}
@@ -2674,7 +2674,7 @@ where
/// budget. Sent directly over the bus; delivery failure is terminal for
/// this reply (the client recovers via its own read-timeout).
#[allow(clippy::future_not_send)]
- async fn deny_partition_request_transient(&self, request_header: &RequestHeader) {
+ async fn deny_partition_request_transient(&self, request_header: &RoutedRequestHeader) {
let reply = build_deny_reply_from_request_header(
request_header,
IggyError::TransientNotAccepted.as_code(),
@@ -2711,7 +2711,7 @@ where
///
/// Returns whether the pump took it. A shard with no sender stages nothing,
/// so assuming success logs an answer for a request destroyed unanswered.
- fn stage_transient_deny(&self, request_header: &RequestHeader) -> bool {
+ fn stage_transient_deny(&self, request_header: &RoutedRequestHeader) -> bool {
let reply = build_deny_reply_from_request_header(
request_header,
IggyError::TransientNotAccepted.as_code(),
@@ -2744,7 +2744,7 @@ where
}
#[allow(clippy::future_not_send)]
- pub async fn on_request(&self, request: Message<RequestHeader>)
+ pub async fn on_request(&self, request: Message<RoutedRequestHeader>)
where
B: MessageBus,
MJ: JournalHandle,
@@ -2959,7 +2959,7 @@ where
return None;
};
debug_assert_eq!(
- partition.consensus().namespace(),
+ partition.consensus().group(),
namespace,
"keyed partition lookup must match the frame namespace"
);
@@ -2984,7 +2984,7 @@ where
let planes = self.plane.inner();
if let Some(ref consensus) = planes.0.consensus
- && consensus.namespace() == header.namespace
+ && consensus.group() == header.group
{
let actions = consensus.handle_start_view_change(PlaneKind::Metadata, &header);
let (local_actions, wire_actions) = split_local_actions(actions);
@@ -2997,7 +2997,7 @@ where
let Some(partition) = self.resolve_partition_target(
&planes.1.0,
- header.namespace,
+ header.group,
header.view,
header.replica,
"StartViewChange",
@@ -3033,7 +3033,7 @@ where
let planes = self.plane.inner();
if let Some(ref consensus) = planes.0.consensus
- && consensus.namespace() == header.namespace
+ && consensus.group() == header.group
{
let actions = consensus.handle_do_view_change(PlaneKind::Metadata, &header);
let (local_actions, wire_actions) = split_local_actions(actions);
@@ -3056,7 +3056,7 @@ where
let config = planes.1.0.config();
let Some(partition) = self.resolve_partition_target(
&planes.1.0,
- header.namespace,
+ header.group,
header.view,
header.replica,
"DoViewChange",
@@ -3103,7 +3103,7 @@ where
let planes = self.plane.inner();
if let Some(ref consensus) = planes.0.consensus
- && consensus.namespace() == header.namespace
+ && consensus.group() == header.group
{
let actions = consensus.handle_start_view(PlaneKind::Metadata, &header);
// Every rejection path (wrong primary, old view, stale incarnation,
@@ -3185,7 +3185,7 @@ where
let transfers_inflight = self.partition_transfers_inflight();
let Some(partition) = self.resolve_partition_target(
&planes.1.0,
- header.namespace,
+ header.group,
header.view,
header.replica,
"StartView",
@@ -3213,7 +3213,7 @@ where
{
tracing::info!(
shard = self.id,
- namespace_raw = header.namespace,
+ namespace_raw = header.group,
peer = header.replica,
"adopted a live view while awaiting transfer; requesting partition state transfer"
);
@@ -3264,7 +3264,7 @@ where
nonce,
from_op,
to_op,
- header.namespace,
+ header.group,
)
.await;
}
@@ -3286,7 +3286,7 @@ where
let planes = self.plane.inner();
if let Some(ref consensus) = planes.0.consensus
- && consensus.namespace() == header.namespace
+ && consensus.group() == header.group
{
match consensus.handle_commit(&header) {
CommitOutcome::Advanced => {
@@ -3334,7 +3334,7 @@ where
let config = planes.1.0.config();
let Some(partition) = self.resolve_partition_target(
&planes.1.0,
- header.namespace,
+ header.group,
header.view,
header.replica,
"Commit",
@@ -3388,7 +3388,7 @@ where
let header = *msg.header();
let planes = self.plane.inner();
if let Some(ref consensus) = planes.0.consensus
- && consensus.namespace() == header.namespace
+ && consensus.group() == header.group
{
let actions = consensus.handle_request_start_view(PlaneKind::Metadata, &header);
let (local_actions, wire_actions) = split_local_actions(actions);
@@ -3401,7 +3401,7 @@ where
let Some(partition) = planes
.1
.0
- .get_mut_by_ns(&IggyNamespace::from_raw(header.namespace))
+ .get_mut_by_ns(&IggyNamespace::from_raw(header.group))
else {
return;
};
@@ -3444,7 +3444,7 @@ where
let repair_chunk_max = self.repair_chunk_max.get();
let planes = self.plane.inner();
if let Some(ref consensus) = planes.0.consensus
- && consensus.namespace() == header.namespace
+ && consensus.group() == header.group
{
if !consensus.is_normal() {
return;
@@ -3476,7 +3476,7 @@ where
Command2::RangeEvicted,
header.nonce,
from_op,
- header.namespace,
+ header.group,
)
.await;
self.send_repair_range_reply(
@@ -3486,7 +3486,7 @@ where
Command2::RepairDone,
header.nonce,
header.from_op.saturating_sub(1),
- header.namespace,
+ header.group,
)
.await;
return;
@@ -3499,7 +3499,7 @@ where
Command2::RangeEvicted,
header.nonce,
from_op,
- header.namespace,
+ header.group,
)
.await;
}
@@ -3528,7 +3528,7 @@ where
Command2::RepairDone,
header.nonce,
served_through,
- header.namespace,
+ header.group,
)
.await;
return;
@@ -3536,7 +3536,7 @@ where
let Some(partition) = planes
.1
.0
- .get_mut_by_ns(&IggyNamespace::from_raw(header.namespace))
+ .get_mut_by_ns(&IggyNamespace::from_raw(header.group))
else {
return;
};
@@ -3558,7 +3558,7 @@ where
Command2::RangeEvicted,
header.nonce,
retained_from,
- header.namespace,
+ header.group,
)
.await;
from_op = retained_from;
@@ -3581,12 +3581,12 @@ where
Command2::RepairDone,
header.nonce,
served_through,
- header.namespace,
+ header.group,
)
.await;
tracing::info!(
shard = self.id,
- namespace_raw = header.namespace,
+ namespace_raw = header.group,
target,
from_op = header.from_op,
to_op,
@@ -3612,7 +3612,7 @@ where
tracing::debug!(
shard = self.id,
op = msg.header().0.op,
- namespace_raw = msg.header().0.namespace,
+ namespace_raw = msg.header().0.group,
"repair prepare received"
);
// Convert to a live-prepare frame exactly once, here at the apply
@@ -3628,15 +3628,15 @@ where
let header = *msg.header();
let planes = self.plane.inner();
// `header.namespace == 0`: pre-upgrade metadata WAL entries were
- // journaled before prepares stamped `consensus.namespace()`, and
+ // journaled before prepares stamped `consensus.group()`, and
// repair ships stored bytes verbatim -- without this acceptance a
// mixed-version metadata repair re-ships the same 0-stamped entries
// forever. Safe to claim for the metadata plane: partition ops never
// reach this arm with a metadata operation code, and namespace 0's
// partition-side gates journal nothing for foreign operations.
if let Some(ref consensus) = planes.0.consensus
- && (consensus.namespace() == header.namespace
- || (header.namespace == 0 && !header.operation.is_partition()))
+ && (consensus.group() == header.group
+ || (header.group == 0 && !header.operation.is_partition()))
{
let session = *self.metadata_repair.borrow();
let Some(session) = session else {
@@ -3680,7 +3680,7 @@ where
let Some(partition) = planes
.1
.0
- .get_mut_by_ns(&IggyNamespace::from_raw(header.namespace))
+ .get_mut_by_ns(&IggyNamespace::from_raw(header.group))
else {
return;
};
@@ -3705,7 +3705,7 @@ where
let header = *msg.header();
let planes = self.plane.inner();
if let Some(ref consensus) = planes.0.consensus
- && consensus.namespace() == header.namespace
+ && consensus.group() == header.group
{
let session = *self.metadata_repair.borrow();
let Some(session) = session else {
@@ -3745,7 +3745,7 @@ where
session.nonce,
commit_min + 1,
session.to_op,
- header.namespace,
+ header.group,
)
.await;
}
@@ -3802,7 +3802,7 @@ where
let Some(partition) = planes
.1
.0
- .get_mut_by_ns(&IggyNamespace::from_raw(header.namespace))
+ .get_mut_by_ns(&IggyNamespace::from_raw(header.group))
else {
return;
};
@@ -3843,7 +3843,7 @@ where
// arming here would defeat its backoff.
tracing::info!(
shard = self.id,
- namespace_raw = header.namespace,
+ namespace_raw = header.group,
floor,
to_op,
peer = header.replica,
@@ -3860,7 +3860,7 @@ where
// the scheduled re-arm) owns recovery from here.
tracing::info!(
shard = self.id,
- namespace_raw = header.namespace,
+ namespace_raw = header.group,
floor,
to_op,
"partition repair floor refused; transfer in flight or scheduled"
@@ -3871,7 +3871,7 @@ where
if partition.repair.is_none() {
tracing::info!(
shard = self.id,
- namespace_raw = header.namespace,
+ namespace_raw = header.group,
through_op = header.op,
"partition journal repair complete"
);
@@ -3890,7 +3890,7 @@ where
nonce,
commit_min + 1,
to_op,
- header.namespace,
+ header.group,
)
.await;
}
@@ -3926,7 +3926,7 @@ where
h.nonce = nonce;
h.from_op = from_op;
h.to_op = to_op;
- h.namespace = namespace;
+ h.group = namespace;
h.size = size_of::<RequestPreparesHeader>() as u32;
});
if self
@@ -4000,7 +4000,7 @@ where
h.replica = self_id;
h.nonce = nonce;
h.op = op;
- h.namespace = namespace;
+ h.group = namespace;
h.size = size_of::<RepairRangeReplyHeader>() as u32;
});
let _ = self
@@ -4045,7 +4045,7 @@ where
nonce,
from_op,
to_op,
- consensus.namespace(),
+ consensus.group(),
)
.await;
}
@@ -4068,7 +4068,7 @@ where
h.cluster = consensus.cluster();
h.replica = consensus.replica();
h.nonce = nonce;
- h.namespace = consensus.namespace();
+ h.group = consensus.group();
h.size = size_of::<RequestStateTransferHeader>() as u32;
});
let _ = self
@@ -4109,7 +4109,7 @@ where
h.cluster = cluster;
h.replica = self_id;
h.nonce = nonce;
- h.namespace = namespace;
+ h.group = namespace;
h.size = total_size as u32;
if let Some((_, commit_op)) = offer {
h.available = 1;
@@ -4146,7 +4146,7 @@ where
h.cluster = cluster;
h.replica = self_id;
h.nonce = nonce;
- h.namespace = namespace;
+ h.group = namespace;
h.artifact = artifact;
h.offset = offset;
h.len = len;
@@ -4178,7 +4178,7 @@ where
.0
.consensus
.as_ref()
- .is_some_and(|consensus| consensus.namespace() == header.namespace);
+ .is_some_and(|consensus| consensus.group() == header.group);
if !metadata_frame {
return self.on_partition_request_state_transfer(msg).await;
}
@@ -4198,7 +4198,7 @@ where
let cached = self
.state_transfer_offers
.borrow_mut()
- .get_mut(&(header.namespace, header.replica))
+ .get_mut(&(header.group, header.replica))
.filter(|served| served.nonce == header.nonce)
.and_then(|served| {
let ServedOffer::Metadata(offer) = &served.offer else {
@@ -4222,7 +4222,7 @@ where
self_id,
header.replica,
header.nonce,
- header.namespace,
+ header.group,
Some((&offer.manifest(), offer.commit_op)),
)
.await;
@@ -4245,12 +4245,12 @@ where
self_id,
header.replica,
header.nonce,
- header.namespace,
+ header.group,
Some((&offer.manifest(), offer.commit_op)),
)
.await;
self.state_transfer_offers.borrow_mut().insert(
- (header.namespace, header.replica),
+ (header.group, header.replica),
ServedStateTransfer {
nonce: header.nonce,
offer: ServedOffer::Metadata(offer),
@@ -4275,7 +4275,7 @@ where
self_id,
header.replica,
header.nonce,
- header.namespace,
+ header.group,
None,
)
.await;
@@ -4316,7 +4316,7 @@ where
.0
.consensus
.as_ref()
- .is_some_and(|consensus| consensus.namespace() == header.namespace);
+ .is_some_and(|consensus| consensus.group() == header.group);
if !metadata_frame {
return self.on_partition_state_transfer_target(msg).await;
}
@@ -4494,7 +4494,7 @@ where
consensus.replica(),
peer,
nonce,
- consensus.namespace(),
+ consensus.group(),
artifact,
offset,
len,
@@ -4604,7 +4604,7 @@ where
.0
.consensus
.as_ref()
- .is_some_and(|consensus| consensus.namespace() == header.namespace);
+ .is_some_and(|consensus| consensus.group() == header.group);
if !metadata_frame {
return self.on_partition_request_state_chunk(msg).await;
}
@@ -4626,7 +4626,7 @@ where
let reply = {
let mut offers = self.state_transfer_offers.borrow_mut();
let served = offers
- .get_mut(&(header.namespace, header.replica))
+ .get_mut(&(header.group, header.replica))
.filter(|served| served.nonce == header.nonce);
served.map_or(Some(ChunkReply::UnknownOffer), |served| {
let ServedOffer::Metadata(offer) = &served.offer else {
@@ -4674,7 +4674,7 @@ where
h.cluster = cluster;
h.replica = self_id;
h.nonce = header.nonce;
- h.namespace = header.namespace;
+ h.group = header.group;
h.artifact = header.artifact;
h.offset = header.offset;
h.size = total_size as u32;
@@ -4700,7 +4700,7 @@ where
self_id,
header.replica,
header.nonce,
- header.namespace,
+ header.group,
None,
)
.await;
@@ -4738,7 +4738,7 @@ where
.0
.consensus
.as_ref()
- .is_some_and(|consensus| consensus.namespace() == header.namespace);
+ .is_some_and(|consensus| consensus.group() == header.group);
if !metadata_frame {
return self.on_partition_state_chunk(msg).await;
}
@@ -5271,7 +5271,7 @@ where
let Some(partition) = planes
.1
.0
- .get_mut_by_ns(&IggyNamespace::from_raw(header.namespace))
+ .get_mut_by_ns(&IggyNamespace::from_raw(header.group))
else {
return;
};
@@ -5286,7 +5286,7 @@ where
let cached = self
.state_transfer_offers
.borrow_mut()
- .get_mut(&(header.namespace, header.replica))
+ .get_mut(&(header.group, header.replica))
.filter(|served| served.nonce == header.nonce)
.and_then(|served| {
let ServedOffer::Partition { offer, .. } = &served.offer else {
@@ -5298,7 +5298,7 @@ where
if let Some(offer) = cached {
tracing::debug!(
shard = self.id,
- namespace_raw = header.namespace,
+ namespace_raw = header.group,
requester = header.replica,
"re-answering a partition state transfer request from the offer already served"
);
@@ -5307,7 +5307,7 @@ where
self_id,
header.replica,
header.nonce,
- header.namespace,
+ header.group,
Some((&offer.manifest(), offer.commit_op)),
)
.await;
@@ -5318,7 +5318,7 @@ where
Ok(offer) => {
tracing::info!(
shard = self.id,
- namespace_raw = header.namespace,
+ namespace_raw = header.group,
requester = header.replica,
commit_op = offer.commit_op,
artifacts = offer.artifact_count(),
@@ -5330,12 +5330,12 @@ where
self_id,
header.replica,
header.nonce,
- header.namespace,
+ header.group,
Some((&offer.manifest(), offer.commit_op)),
)
.await;
self.state_transfer_offers.borrow_mut().insert(
- (header.namespace, header.replica),
+ (header.group, header.replica),
ServedStateTransfer {
nonce: header.nonce,
offer: ServedOffer::Partition { offer },
@@ -5350,7 +5350,7 @@ where
// operator-visible fault on THIS node.
tracing::info!(
shard = self.id,
- namespace_raw = header.namespace,
+ namespace_raw = header.group,
requester = header.replica,
%reason,
"cannot serve partition state transfer; requester falls back"
@@ -5360,7 +5360,7 @@ where
self_id,
header.replica,
header.nonce,
- header.namespace,
+ header.group,
None,
)
.await;
@@ -5397,11 +5397,7 @@ where
let header = *msg.header();
let planes = self.plane.inner();
- let Some(partition) = planes
- .1
- .0
- .get_by_ns(&IggyNamespace::from_raw(header.namespace))
- else {
+ let Some(partition) = planes.1.0.get_by_ns(&IggyNamespace::from_raw(header.group)) else {
return;
};
let cluster = partition.consensus().cluster();
@@ -5411,7 +5407,7 @@ where
let attempt = 'attempt: {
let mut offers = self.state_transfer_offers.borrow_mut();
let served = offers
- .get_mut(&(header.namespace, header.replica))
+ .get_mut(&(header.group, header.replica))
.filter(|served| served.nonce == header.nonce);
let Some(served) = served else {
break 'attempt ChunkAttempt::Reply(Some(ChunkReply::UnknownOffer));
@@ -5431,7 +5427,7 @@ where
match self
.served_segment_cache
.borrow_mut()
- .get(header.namespace, source.entry.checksum)
+ .get(header.group, source.entry.checksum)
{
Some(payload) => {
segment_payload = payload;
@@ -5473,7 +5469,7 @@ where
h.cluster = cluster;
h.replica = self_id;
h.nonce = header.nonce;
- h.namespace = header.namespace;
+ h.group = header.group;
h.artifact = header.artifact;
h.offset = header.offset;
h.size = total_size as u32;
@@ -5485,7 +5481,7 @@ where
ChunkAttempt::Load { log_path, entry } => {
if let Some(bytes) = load_partition_artifact(&log_path, &entry).await {
self.served_segment_cache.borrow_mut().insert(
- header.namespace,
+ header.group,
entry.checksum,
Rc::new(bytes),
);
@@ -5493,7 +5489,7 @@ where
}
tracing::warn!(
shard = self.id,
- namespace_raw = header.namespace,
+ namespace_raw = header.group,
artifact = header.artifact,
path = %log_path,
"served segment no longer matches its manifest entry; \
@@ -5501,7 +5497,7 @@ where
);
self.state_transfer_offers
.borrow_mut()
- .remove(&(header.namespace, header.replica));
+ .remove(&(header.group, header.replica));
// The builder cache too: it is keyed by commit_op alone,
// and GC unlinks files WITHOUT a commit, so the restarted
// requester would otherwise be handed the same offer with
@@ -5521,7 +5517,7 @@ where
Some(ChunkReply::UnknownOffer) => {
tracing::info!(
shard = self.id,
- namespace_raw = header.namespace,
+ namespace_raw = header.group,
requester = header.replica,
"partition chunk request for an unknown offer; telling requester to restart"
);
@@ -5530,7 +5526,7 @@ where
self_id,
header.replica,
header.nonce,
- header.namespace,
+ header.group,
None,
)
.await;
@@ -5538,7 +5534,7 @@ where
None => {
tracing::warn!(
shard = self.id,
- namespace_raw = header.namespace,
+ namespace_raw = header.group,
requester = header.replica,
artifact = header.artifact,
offset = header.offset,
@@ -5613,7 +5609,7 @@ where
{
tracing::info!(
shard = self.id,
- namespace_raw = partition.consensus().namespace(),
+ namespace_raw = partition.consensus().group(),
cap = Self::PARTITION_TRANSFERS_INFLIGHT_MAX,
"partition transfer slots exhausted; deferring this arm"
);
@@ -5663,7 +5659,7 @@ where
let to_op = consensus.commit_max();
let cluster = consensus.cluster();
let self_id = consensus.replica();
- let namespace = consensus.namespace();
+ let namespace = consensus.group();
partition.repair = Some(partitions::RepairSession {
nonce,
to_op,
@@ -5697,7 +5693,7 @@ where
let Some(partition) = planes
.1
.0
- .get_mut_by_ns(&IggyNamespace::from_raw(header.namespace))
+ .get_mut_by_ns(&IggyNamespace::from_raw(header.group))
else {
return;
};
@@ -5717,7 +5713,7 @@ where
// shared re-arm path.
tracing::info!(
shard = self.id,
- namespace_raw = header.namespace,
+ namespace_raw = header.group,
peer = header.replica,
"partition transfer peer cannot serve; backing off before re-arming"
);
@@ -5732,7 +5728,7 @@ where
Err(error) => {
tracing::warn!(
shard = self.id,
- namespace_raw = header.namespace,
+ namespace_raw = header.group,
%error,
"partition transfer manifest rejected"
);
@@ -5759,7 +5755,7 @@ where
if !kind_capped || total_len > Self::PARTITION_TRANSFER_TOTAL_LEN_MAX {
tracing::warn!(
shard = self.id,
- namespace_raw = header.namespace,
+ namespace_raw = header.group,
total_len,
"partition transfer manifest exceeds artifact caps; refusing descriptor"
);
@@ -5808,7 +5804,7 @@ where
if consensus.state_transfer_stage() == consensus::StateTransferStage::AwaitingTarget {
consensus.set_state_transfer_stage(consensus::StateTransferStage::Fetching);
}
- self.on_partition_transfer_progress(header.namespace).await;
+ self.on_partition_transfer_progress(header.group).await;
}
/// Receive one partition chunk; spill a completed segment artifact, and
@@ -5824,7 +5820,7 @@ where
let Some(partition) = planes
.1
.0
- .get_mut_by_ns(&IggyNamespace::from_raw(header.namespace))
+ .get_mut_by_ns(&IggyNamespace::from_raw(header.group))
else {
return;
};
@@ -5847,7 +5843,7 @@ where
session.idle_ticks = 0;
}
partition.note_transfer_progress();
- self.on_partition_transfer_progress(header.namespace).await;
+ self.on_partition_transfer_progress(header.group).await;
}
/// Drive an in-flight partition transfer: spill newly completed segment
@@ -6114,7 +6110,7 @@ where
let after_ticks = transfer_rearm_backoff(self.repair_retry_ticks.get(), failures);
tracing::info!(
shard = self.id,
- namespace_raw = partition.consensus().namespace(),
+ namespace_raw = partition.consensus().group(),
failures,
next_peer,
after_ticks,
@@ -6212,7 +6208,7 @@ where
let metadata_served = metadata.consensus.as_ref().is_some_and(|consensus| {
offers
.keys()
- .any(|(namespace, _)| *namespace == consensus.namespace())
+ .any(|(namespace, _)| *namespace == consensus.group())
});
if !metadata_served {
metadata.clear_state_transfer_offer_cache();
@@ -6369,7 +6365,7 @@ where
nonce,
from_op,
to_op,
- consensus.namespace(),
+ consensus.group(),
)
.await;
}
@@ -6395,7 +6391,7 @@ where
view = consensus.view(),
op = consensus.sequencer().current_sequence(),
commit = consensus.commit_max(),
- namespace = consensus.namespace(),
+ namespace = consensus.group(),
"answering stale-view heartbeat with StartView"
);
// Unsolicited, answering a stale-view heartbeat rather than a probe, so there is
@@ -6408,7 +6404,7 @@ where
commit: consensus.commit_max(),
incarnation: 0,
target: None,
- namespace: consensus.namespace(),
+ group: consensus.group(),
};
dispatch_vsr_actions::<B, P, J>(consensus, None, &[action]).await;
}
@@ -6524,21 +6520,21 @@ async fn dispatch_vsr_actions<B, P, J>(
!advertises_view || !consensus.needs_superblock_persist(),
"durable-before-send violated: dispatching a view-scoped action for \
namespace {} while the superblock is behind the in-memory view {}",
- consensus.namespace(),
+ consensus.group(),
consensus.view(),
);
}
for action in actions {
match action {
- VsrAction::SendStartViewChange { view, namespace } => {
+ VsrAction::SendStartViewChange { view, group } => {
let msg = Message::<StartViewChangeHeader>::new(size_of::<StartViewChangeHeader>())
.transmute_header(|_, h: &mut StartViewChangeHeader| {
h.command = Command2::StartViewChange;
h.cluster = cluster;
h.replica = self_id;
h.view = *view;
- h.namespace = *namespace;
+ h.group = *group;
h.size = size_of::<StartViewChangeHeader>() as u32;
});
broadcast(msg.into_generic().into_frozen()).await;
@@ -6549,7 +6545,7 @@ async fn dispatch_vsr_actions<B, P, J>(
log_view,
op,
commit,
- namespace,
+ group,
} => {
let msg = Message::<DoViewChangeHeader>::new(size_of::<DoViewChangeHeader>())
.transmute_header(|_, h: &mut DoViewChangeHeader| {
@@ -6560,12 +6556,12 @@ async fn dispatch_vsr_actions<B, P, J>(
h.log_view = *log_view;
h.op = *op;
h.commit = *commit;
- h.namespace = *namespace;
+ h.group = *group;
h.size = size_of::<DoViewChangeHeader>() as u32;
});
send(*target, msg.into_generic().into_frozen()).await;
}
- VsrAction::SendRequestStartView { view, namespace } => {
+ VsrAction::SendRequestStartView { view, group } => {
// Stamp this replica's incarnation so the answering StartView can
// echo it, proving to us the reply post-dates our restart.
let incarnation = consensus.incarnation();
@@ -6577,7 +6573,7 @@ async fn dispatch_vsr_actions<B, P, J>(
h.replica = self_id;
h.view = *view;
h.incarnation = incarnation;
- h.namespace = *namespace;
+ h.group = *group;
h.size = size_of::<RequestStartViewHeader>() as u32;
});
broadcast(msg.into_generic().into_frozen()).await;
@@ -6588,7 +6584,7 @@ async fn dispatch_vsr_actions<B, P, J>(
commit,
incarnation,
target,
- namespace,
+ group,
} => {
let msg = Message::<StartViewHeader>::new(size_of::<StartViewHeader>())
.transmute_header(|_, h: &mut StartViewHeader| {
@@ -6599,7 +6595,7 @@ async fn dispatch_vsr_actions<B, P, J>(
h.op = *op;
h.commit = *commit;
h.incarnation = *incarnation;
- h.namespace = *namespace;
+ h.group = *group;
h.size = size_of::<StartViewHeader>() as u32;
});
let frozen = msg.into_generic().into_frozen();
@@ -6617,7 +6613,7 @@ async fn dispatch_vsr_actions<B, P, J>(
from_op,
to_op,
target,
- namespace,
+ group,
} => {
let Some(journal) = journal else {
continue;
@@ -6640,7 +6636,7 @@ async fn dispatch_vsr_actions<B, P, J>(
h.prepare_checksum = prepare_header.checksum;
h.request = prepare_header.request;
h.operation = prepare_header.operation;
- h.namespace = *namespace;
+ h.group = *group;
h.size = size_of::<PrepareOkHeader>() as u32;
});
send(*target, msg.into_generic().into_frozen()).await;
@@ -6725,7 +6721,7 @@ async fn dispatch_vsr_actions<B, P, J>(
VsrAction::SendCommit {
view,
commit,
- namespace,
+ group,
timestamp_monotonic,
} => {
let msg = Message::<CommitHeader>::new(size_of::<CommitHeader>()).transmute_header(
@@ -6735,7 +6731,7 @@ async fn dispatch_vsr_actions<B, P, J>(
h.replica = self_id;
h.view = *view;
h.commit = *commit;
- h.namespace = *namespace;
+ h.group = *group;
h.timestamp_monotonic = *timestamp_monotonic;
h.size = size_of::<CommitHeader>() as u32;
},
@@ -6783,7 +6779,7 @@ async fn dispatch_partition_journal_actions<B, P, SB>(
|| !consensus.needs_superblock_persist(),
"durable-before-send violated: dispatching a view-scoped action for \
namespace {} while the superblock is behind the in-memory view {}",
- consensus.namespace(),
+ consensus.group(),
consensus.view(),
);
}
@@ -6795,7 +6791,7 @@ async fn dispatch_partition_journal_actions<B, P, SB>(
from_op,
to_op,
target,
- namespace,
+ group,
} => {
for op in *from_op..=*to_op {
let Some(prepare_header) = journal.header_by_op(op) else {
@@ -6814,7 +6810,7 @@ async fn dispatch_partition_journal_actions<B, P, SB>(
h.prepare_checksum = prepare_header.checksum;
h.request = prepare_header.request;
h.operation = prepare_header.operation;
- h.namespace = *namespace;
+ h.group = *group;
h.size = size_of::<PrepareOkHeader>() as u32;
});
send(*target, msg.into_generic().into_frozen()).await;
@@ -6926,7 +6922,7 @@ mod persist_gate_tests {
commit: 3,
incarnation: 0,
target: None,
- namespace: 7,
+ group: 7,
},
VsrAction::CommitJournal,
rebuild(),
@@ -6946,10 +6942,7 @@ mod persist_gate_tests {
#[test]
fn given_send_only_actions_when_split_should_leave_locals_empty() {
- let actions = vec![VsrAction::SendStartViewChange {
- view: 2,
- namespace: 7,
- }];
+ let actions = vec![VsrAction::SendStartViewChange { view: 2, group: 7 }];
let (local, wire) = split_local_actions(actions);
assert!(local.is_empty());
assert_eq!(wire.len(), 1);
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index 90eb71823..c5ecff6b3 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -27,7 +27,7 @@ use iggy_binary_protocol::{ConsensusHeader, GenericHeader, Operation, PrepareHea
use journal::superblock::SuperblockStore;
use journal::{Journal, JournalHandle};
use message_bus::{ConnectionInstaller, MessageBus, ReplicaHandshakeDoneFn};
-use server_common::sharding::{IggyNamespace, METADATA_CONSENSUS_NAMESPACE};
+use server_common::sharding::{IggyNamespace, METADATA_GROUP};
use server_common::{Message, MessageBag};
/// How often the shard pump drives `VsrConsensus::tick`.
@@ -46,63 +46,63 @@ fn extract_routing(bag: MessageBag) -> (Operation, u64, Message<GenericHeader>)
match bag {
MessageBag::Request(r) => {
let h = *r.header();
- (h.operation, h.namespace, r.into_generic())
+ (h.operation, h.group, r.into_generic())
}
MessageBag::Prepare(p) => {
let h = *p.header();
- (h.operation, h.namespace, p.into_generic())
+ (h.operation, h.group, p.into_generic())
}
MessageBag::PrepareOk(p) => {
let h = *p.header();
- (h.operation, h.namespace, p.into_generic())
+ (h.operation, h.group, p.into_generic())
}
MessageBag::StartViewChange(m) => {
let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
+ (h.operation(), h.group, m.into_generic())
}
MessageBag::DoViewChange(m) => {
let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
+ (h.operation(), h.group, m.into_generic())
}
MessageBag::StartView(m) => {
let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
+ (h.operation(), h.group, m.into_generic())
}
MessageBag::Commit(m) => {
let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
+ (h.operation(), h.group, m.into_generic())
}
MessageBag::RequestStartView(m) => {
let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
+ (h.operation(), h.group, m.into_generic())
}
MessageBag::RequestPrepares(m) => {
let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
+ (h.operation(), h.group, m.into_generic())
}
MessageBag::RepairPrepare(m) => {
let h = *m.header();
- (h.0.operation, h.0.namespace, m.into_generic())
+ (h.0.operation, h.0.group, m.into_generic())
}
MessageBag::RepairRangeReply(m) => {
let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
+ (h.operation(), h.group, m.into_generic())
}
MessageBag::RequestStateTransfer(m) => {
let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
+ (h.operation(), h.group, m.into_generic())
}
MessageBag::StateTransferTarget(m) => {
let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
+ (h.operation(), h.group, m.into_generic())
}
MessageBag::RequestStateChunk(m) => {
let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
+ (h.operation(), h.group, m.into_generic())
}
MessageBag::StateChunk(m) => {
let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
+ (h.operation(), h.group, m.into_generic())
}
}
}
@@ -185,7 +185,7 @@ where
/// frame (`StartViewChange`, `DoViewChange`, `StartView`, `Commit`)
/// or a client `Register` request. The owning consensus group is
/// identified by `namespace_u64`:
- /// - `METADATA_CONSENSUS_NAMESPACE` -> shard 0.
+ /// - `METADATA_GROUP` -> shard 0.
/// - packable `IggyNamespace::inner()` -> the shard owning that
/// partition's consensus group.
fn route_typed(
@@ -232,7 +232,7 @@ where
"route_typed: operation {operation:?} fell through unclassified; \
expected is_metadata / is_partition / is_vsr_reserved"
);
- if namespace_u64 == METADATA_CONSENSUS_NAMESPACE {
+ if namespace_u64 == METADATA_GROUP {
self.try_send_to_target(0, generic, operation);
return;
}
diff --git a/core/simulator/src/client.rs b/core/simulator/src/client.rs
index 028adb38c..02ea37323 100644
--- a/core/simulator/src/client.rs
+++ b/core/simulator/src/client.rs
@@ -43,12 +43,12 @@ use iggy_binary_protocol::requests::users::{
UpdatePermissionsRequest, UpdateUserRequest,
};
use iggy_binary_protocol::{
- AckLevel, ClientVersionInfo, IGGY_PROTOCOL_VERSION, Operation, RequestHeader, WireEncode,
+ AckLevel, ClientVersionInfo, IGGY_PROTOCOL_VERSION, Operation, RoutedRequestHeader, WireEncode,
WireIdentifier, WireName, WirePartitioning, WirePollingStrategy,
};
use metadata::stm::user::{CreatePersonalAccessTokenRequest, DeletePersonalAccessTokenRequest};
use secrecy::SecretString;
-use server_common::sharding::{IggyNamespace, METADATA_CONSENSUS_NAMESPACE};
+use server_common::sharding::{IggyNamespace, METADATA_GROUP};
use server_common::{Message, iobuf::Owned};
use std::cell::Cell;
@@ -130,7 +130,7 @@ impl SimClient {
/// offset into a disjoint range ([`PARTITION_ID_BASE`]). A partition id can
/// therefore never equal a metadata id, so a delayed or duplicated partition
/// reply is never misattributed to a metadata entry in the auditor's
- /// `(client, request)` map (which would trip the namespace guard and drop a
+ /// `(client, request)` map (which would trip the group guard and drop a
/// live metadata op). This holds regardless of reply duplication, not only
/// while clients are one-in-flight.
fn request_id_for(&self, operation: Operation) -> u64 {
@@ -162,9 +162,9 @@ impl SimClient {
/// # Panics
/// Panics if the register request buffer is invalid.
#[allow(clippy::cast_possible_truncation)]
- pub fn register(&self) -> Message<RequestHeader> {
- let header_size = std::mem::size_of::<RequestHeader>();
- let header = RequestHeader {
+ pub fn register(&self) -> Message<RoutedRequestHeader> {
+ let header_size = std::mem::size_of::<RoutedRequestHeader>();
+ let header = RoutedRequestHeader {
command: iggy_binary_protocol::Command2::Request,
operation: Operation::Register,
size: header_size as u32,
@@ -173,8 +173,8 @@ impl SimClient {
request: 0,
// Register is a vsr-reserved op: the shard router picks its
// target by comparing this against the metadata consensus
- // namespace, not by op class.
- namespace: METADATA_CONSENSUS_NAMESPACE,
+ // group, not by op class.
+ group: METADATA_GROUP,
..Default::default()
};
@@ -198,7 +198,7 @@ impl SimClient {
/// Panics if a credential exceeds the wire name/secret bounds or the
/// request buffer is invalid.
#[allow(clippy::cast_possible_truncation)]
- pub fn login(&self, username: &str, password: &str) -> Message<RequestHeader> {
+ pub fn login(&self, username: &str, password: &str) -> Message<RoutedRequestHeader> {
let body = LoginRegisterRequest {
version_info: ClientVersionInfo {
protocol_version: IGGY_PROTOCOL_VERSION,
@@ -211,16 +211,16 @@ impl SimClient {
}
.to_bytes();
- let header_size = std::mem::size_of::<RequestHeader>();
+ let header_size = std::mem::size_of::<RoutedRequestHeader>();
let total_size = header_size + body.len();
- let header = RequestHeader {
+ let header = RoutedRequestHeader {
command: iggy_binary_protocol::Command2::Request,
operation: Operation::Register,
size: total_size as u32,
client: self.client_id,
session: 0,
request: 0,
- namespace: METADATA_CONSENSUS_NAMESPACE,
+ group: METADATA_GROUP,
..Default::default()
};
@@ -233,7 +233,7 @@ impl SimClient {
/// # Panics
/// Panics if the stream name is not a valid wire name.
- pub fn create_stream(&self, name: &str) -> Message<RequestHeader> {
+ pub fn create_stream(&self, name: &str) -> Message<RoutedRequestHeader> {
let wire = CreateStreamRequest {
name: WireName::new(name).expect("stream name must be valid"),
};
@@ -244,7 +244,7 @@ impl SimClient {
/// # Panics
/// Panics if the stream name cannot be converted to a `WireIdentifier`.
- pub fn delete_stream(&self, name: &str) -> Message<RequestHeader> {
+ pub fn delete_stream(&self, name: &str) -> Message<RoutedRequestHeader> {
let wire = DeleteStreamRequest {
stream_id: WireIdentifier::named(name).expect("stream name must be valid"),
};
@@ -256,7 +256,7 @@ impl SimClient {
/// # Panics
/// Panics if the new name or the existing stream name is not a valid
/// `WireName`.
- pub fn update_stream(&self, stream: &str, new_name: &str) -> Message<RequestHeader> {
+ pub fn update_stream(&self, stream: &str, new_name: &str) -> Message<RoutedRequestHeader> {
let wire = UpdateStreamRequest {
stream_id: WireIdentifier::named(stream).expect("stream name must be valid"),
name: WireName::new(new_name).expect("stream name must be valid"),
@@ -266,7 +266,7 @@ impl SimClient {
/// # Panics
/// Panics if `stream` is not a valid `WireName`.
- pub fn purge_stream(&self, stream: &str) -> Message<RequestHeader> {
+ pub fn purge_stream(&self, stream: &str) -> Message<RoutedRequestHeader> {
let wire = PurgeStreamRequest {
stream_id: WireIdentifier::named(stream).expect("stream name must be valid"),
};
@@ -280,7 +280,7 @@ impl SimClient {
stream: &str,
name: &str,
partitions_count: u32,
- ) -> Message<RequestHeader> {
+ ) -> Message<RoutedRequestHeader> {
let wire = CreateTopicRequest {
stream_id: WireIdentifier::named(stream).expect("stream name must be valid"),
partitions_count,
@@ -300,7 +300,7 @@ impl SimClient {
stream: &str,
topic: &str,
new_name: &str,
- ) -> Message<RequestHeader> {
+ ) -> Message<RoutedRequestHeader> {
let wire = UpdateTopicRequest {
stream_id: WireIdentifier::named(stream).expect("stream name must be valid"),
topic_id: WireIdentifier::named(topic).expect("topic name must be valid"),
@@ -315,7 +315,7 @@ impl SimClient {
/// # Panics
/// Panics if `stream` or `topic` is not a valid `WireName`.
- pub fn delete_topic(&self, stream: &str, topic: &str) -> Message<RequestHeader> {
+ pub fn delete_topic(&self, stream: &str, topic: &str) -> Message<RoutedRequestHeader> {
let wire = DeleteTopicRequest {
stream_id: WireIdentifier::named(stream).expect("stream name must be valid"),
topic_id: WireIdentifier::named(topic).expect("topic name must be valid"),
@@ -325,7 +325,7 @@ impl SimClient {
/// # Panics
/// Panics if `stream` or `topic` is not a valid `WireName`.
- pub fn purge_topic(&self, stream: &str, topic: &str) -> Message<RequestHeader> {
+ pub fn purge_topic(&self, stream: &str, topic: &str) -> Message<RoutedRequestHeader> {
let wire = PurgeTopicRequest {
stream_id: WireIdentifier::named(stream).expect("stream name must be valid"),
topic_id: WireIdentifier::named(topic).expect("topic name must be valid"),
@@ -340,7 +340,7 @@ impl SimClient {
stream: &str,
topic: &str,
partitions_count: u32,
- ) -> Message<RequestHeader> {
+ ) -> Message<RoutedRequestHeader> {
let wire = CreatePartitionsRequest {
stream_id: WireIdentifier::named(stream).expect("stream name must be valid"),
topic_id: WireIdentifier::named(topic).expect("topic name must be valid"),
@@ -356,7 +356,7 @@ impl SimClient {
stream: &str,
topic: &str,
partitions_count: u32,
- ) -> Message<RequestHeader> {
+ ) -> Message<RoutedRequestHeader> {
let wire = DeletePartitionsRequest {
stream_id: WireIdentifier::named(stream).expect("stream name must be valid"),
topic_id: WireIdentifier::named(topic).expect("topic name must be valid"),
@@ -373,7 +373,7 @@ impl SimClient {
topic: &str,
partition_id: u32,
segments_count: u32,
- ) -> Message<RequestHeader> {
+ ) -> Message<RoutedRequestHeader> {
let wire = DeleteSegmentsRequest {
stream_id: WireIdentifier::named(stream).expect("stream name must be valid"),
topic_id: WireIdentifier::named(topic).expect("topic name must be valid"),
@@ -390,7 +390,7 @@ impl SimClient {
stream: &str,
topic: &str,
name: &str,
- ) -> Message<RequestHeader> {
+ ) -> Message<RoutedRequestHeader> {
let wire = CreateConsumerGroupRequest {
stream_id: WireIdentifier::named(stream).expect("stream name must be valid"),
topic_id: WireIdentifier::named(topic).expect("topic name must be valid"),
@@ -406,7 +406,7 @@ impl SimClient {
stream: &str,
topic: &str,
group: &str,
- ) -> Message<RequestHeader> {
+ ) -> Message<RoutedRequestHeader> {
let wire = DeleteConsumerGroupRequest {
stream_id: WireIdentifier::named(stream).expect("stream name must be valid"),
topic_id: WireIdentifier::named(topic).expect("topic name must be valid"),
@@ -422,7 +422,7 @@ impl SimClient {
username: &str,
password: &str,
status: u8,
- ) -> Message<RequestHeader> {
+ ) -> Message<RoutedRequestHeader> {
let wire = CreateUserRequest {
username: WireName::new(username).expect("username must be valid"),
password: password.to_string(),
@@ -440,7 +440,7 @@ impl SimClient {
user: &str,
new_username: Option<&str>,
status: Option<u8>,
- ) -> Message<RequestHeader> {
+ ) -> Message<RoutedRequestHeader> {
let wire = UpdateUserRequest {
user_id: WireIdentifier::named(user).expect("username must be valid"),
username: new_username.map(|n| WireName::new(n).expect("username must be valid")),
@@ -451,7 +451,7 @@ impl SimClient {
/// # Panics
/// Panics if `user` is not a valid `WireName`.
- pub fn delete_user(&self, user: &str) -> Message<RequestHeader> {
+ pub fn delete_user(&self, user: &str) -> Message<RoutedRequestHeader> {
let wire = DeleteUserRequest {
user_id: WireIdentifier::named(user).expect("username must be valid"),
};
@@ -465,7 +465,7 @@ impl SimClient {
user: &str,
current_password: &str,
new_password: &str,
- ) -> Message<RequestHeader> {
+ ) -> Message<RoutedRequestHeader> {
let wire = ChangePasswordRequest {
user_id: WireIdentifier::named(user).expect("username must be valid"),
current_password: current_password.to_string(),
@@ -476,7 +476,7 @@ impl SimClient {
/// # Panics
/// Panics if `user` is not a valid `WireName`.
- pub fn update_permissions(&self, user: &str) -> Message<RequestHeader> {
+ pub fn update_permissions(&self, user: &str) -> Message<RoutedRequestHeader> {
let wire = UpdatePermissionsRequest {
user_id: WireIdentifier::named(user).expect("username must be valid"),
permissions: None,
@@ -486,7 +486,11 @@ impl SimClient {
/// # Panics
/// Panics if `name` is not a valid `WireName`.
- pub fn create_personal_access_token(&self, name: &str, expiry: u64) -> Message<RequestHeader> {
+ pub fn create_personal_access_token(
+ &self,
+ name: &str,
+ expiry: u64,
+ ) -> Message<RoutedRequestHeader> {
let wire = CreatePersonalAccessTokenRequest {
user_id: 0,
name: WireName::new(name).expect("PAT name must be valid"),
@@ -502,7 +506,7 @@ impl SimClient {
/// # Panics
/// Panics if `name` is not a valid `WireName`.
- pub fn delete_personal_access_token(&self, name: &str) -> Message<RequestHeader> {
+ pub fn delete_personal_access_token(&self, name: &str) -> Message<RoutedRequestHeader> {
let wire = DeletePersonalAccessTokenRequest {
user_id: 0,
name: WireName::new(name).expect("PAT name must be valid"),
@@ -519,16 +523,16 @@ impl SimClient {
/// path converts it to `SendMessages2` via `transcode_legacy_request`.
///
/// # Panics
- /// Panics if a namespace id exceeds `u32` or the request buffer is invalid.
+ /// Panics if a group id exceeds `u32` or the request buffer is invalid.
pub fn send_messages(
&self,
- namespace: IggyNamespace,
+ group: IggyNamespace,
messages: &[Bytes],
- ) -> Message<RequestHeader> {
- let to_u32 = |v: usize| u32::try_from(v).expect("namespace id fits u32");
- let stream_id = WireIdentifier::Numeric(to_u32(namespace.stream_id()));
- let topic_id = WireIdentifier::Numeric(to_u32(namespace.topic_id()));
- let partitioning = WirePartitioning::PartitionId(to_u32(namespace.partition_id()));
+ ) -> Message<RoutedRequestHeader> {
+ let to_u32 = |v: usize| u32::try_from(v).expect("group id fits u32");
+ let stream_id = WireIdentifier::Numeric(to_u32(group.stream_id()));
+ let topic_id = WireIdentifier::Numeric(to_u32(group.topic_id()));
+ let partitioning = WirePartitioning::PartitionId(to_u32(group.partition_id()));
// Stamp a deterministic, non-zero id per message. `id: 0` (the real
// SDK's server-assigned path) would make the server mint an unseeded
@@ -549,11 +553,11 @@ impl SimClient {
let mut buf = BytesMut::with_capacity(size);
SendMessagesEncoder::encode(&mut buf, &stream_id, &topic_id, &partitioning, &raw);
- self.build_request_with_namespace(Operation::SendMessages, &buf, namespace)
+ self.build_request_with_namespace(Operation::SendMessages, &buf, group)
}
/// Build a `POLL_MESSAGES` request for an individual consumer, reading
- /// `count` messages from offset 0 of `namespace`'s partition.
+ /// `count` messages from offset 0 of `group`'s partition.
///
/// A `NonReplicated` read: the command code sits in the header's
/// `reserved` prefix, and the request id ECHOES the current metadata
@@ -564,8 +568,8 @@ impl SimClient {
/// # Panics
/// Panics if the session is unbound or the request buffer is invalid.
#[allow(clippy::cast_possible_truncation)]
- pub fn poll_messages(&self, namespace: IggyNamespace, count: u32) -> Message<RequestHeader> {
- let (stream_id, topic_id, partition_id) = namespace_ids(namespace);
+ pub fn poll_messages(&self, group: IggyNamespace, count: u32) -> Message<RoutedRequestHeader> {
+ let (stream_id, topic_id, partition_id) = namespace_ids(group);
let body = PollMessagesRequest {
consumer: WireConsumer::consumer(WireIdentifier::Numeric(self.client_id as u32)),
stream_id,
@@ -577,11 +581,11 @@ impl SimClient {
}
.to_bytes();
- let header_size = std::mem::size_of::<RequestHeader>();
+ let header_size = std::mem::size_of::<RoutedRequestHeader>();
let total_size = header_size + body.len();
let mut reserved = [0u8; 52];
reserved[..4].copy_from_slice(&POLL_MESSAGES_CODE.to_le_bytes());
- let header = RequestHeader {
+ let header = RoutedRequestHeader {
command: iggy_binary_protocol::Command2::Request,
operation: Operation::NonReplicated,
size: total_size as u32,
@@ -589,7 +593,7 @@ impl SimClient {
session: self.session_id(),
request: self.request_counter.get(),
reserved,
- namespace: namespace.inner(),
+ group: group.inner(),
..Default::default()
};
@@ -602,12 +606,12 @@ impl SimClient {
pub fn store_consumer_offset(
&self,
- namespace: IggyNamespace,
+ group: IggyNamespace,
consumer_kind: u8,
consumer_id: u32,
offset: u64,
- ) -> Message<RequestHeader> {
- let (stream_id, topic_id, partition_id) = namespace_ids(namespace);
+ ) -> Message<RoutedRequestHeader> {
+ let (stream_id, topic_id, partition_id) = namespace_ids(group);
let request = StoreConsumerOffsetRequest {
consumer: namespace_consumer(consumer_kind, consumer_id),
stream_id,
@@ -618,17 +622,17 @@ impl SimClient {
self.build_request_with_namespace(
Operation::StoreConsumerOffset,
&request.to_bytes(),
- namespace,
+ group,
)
}
pub fn delete_consumer_offset(
&self,
- namespace: IggyNamespace,
+ group: IggyNamespace,
consumer_kind: u8,
consumer_id: u32,
- ) -> Message<RequestHeader> {
- let (stream_id, topic_id, partition_id) = namespace_ids(namespace);
+ ) -> Message<RoutedRequestHeader> {
+ let (stream_id, topic_id, partition_id) = namespace_ids(group);
let request = DeleteConsumerOffsetRequest {
consumer: namespace_consumer(consumer_kind, consumer_id),
stream_id,
@@ -638,7 +642,7 @@ impl SimClient {
self.build_request_with_namespace(
Operation::DeleteConsumerOffset,
&request.to_bytes(),
- namespace,
+ group,
)
}
@@ -647,16 +651,16 @@ impl SimClient {
///
/// # Panics
/// Panics on payload too large for `Owned::<4096>` or invalid
- /// `Message<RequestHeader>` parse; both are simulator misconfig.
+ /// `Message<RoutedRequestHeader>` parse; both are simulator misconfig.
pub fn store_consumer_offset_2(
&self,
- namespace: IggyNamespace,
+ group: IggyNamespace,
consumer_kind: u8,
consumer_id: u32,
offset: u64,
ack: AckLevel,
- ) -> Message<RequestHeader> {
- let (stream_id, topic_id, partition_id) = namespace_ids(namespace);
+ ) -> Message<RoutedRequestHeader> {
+ let (stream_id, topic_id, partition_id) = namespace_ids(group);
let request = StoreConsumerOffset2Request {
consumer: namespace_consumer(consumer_kind, consumer_id),
stream_id,
@@ -668,7 +672,7 @@ impl SimClient {
self.build_request_with_namespace(
Operation::StoreConsumerOffset2,
&request.to_bytes(),
- namespace,
+ group,
)
}
@@ -676,15 +680,15 @@ impl SimClient {
///
/// # Panics
/// Panics on payload too large for `Owned::<4096>` or invalid
- /// `Message<RequestHeader>` parse; both are simulator misconfig.
+ /// `Message<RoutedRequestHeader>` parse; both are simulator misconfig.
pub fn delete_consumer_offset_2(
&self,
- namespace: IggyNamespace,
+ group: IggyNamespace,
consumer_kind: u8,
consumer_id: u32,
ack: AckLevel,
- ) -> Message<RequestHeader> {
- let (stream_id, topic_id, partition_id) = namespace_ids(namespace);
+ ) -> Message<RoutedRequestHeader> {
+ let (stream_id, topic_id, partition_id) = namespace_ids(group);
let request = DeleteConsumerOffset2Request {
consumer: namespace_consumer(consumer_kind, consumer_id),
stream_id,
@@ -695,7 +699,7 @@ impl SimClient {
self.build_request_with_namespace(
Operation::DeleteConsumerOffset2,
&request.to_bytes(),
- namespace,
+ group,
)
}
@@ -703,12 +707,12 @@ impl SimClient {
&self,
operation: Operation,
payload: &[u8],
- namespace: IggyNamespace,
- ) -> Message<RequestHeader> {
- let header_size = std::mem::size_of::<RequestHeader>();
+ group: IggyNamespace,
+ ) -> Message<RoutedRequestHeader> {
+ let header_size = std::mem::size_of::<RoutedRequestHeader>();
let total_size = header_size + payload.len();
- let header = self.header(operation, namespace.inner(), total_size);
+ let header = self.header(operation, group.inner(), total_size);
let header_bytes = bytemuck::bytes_of(&header);
let mut buffer = Vec::with_capacity(total_size);
@@ -719,14 +723,14 @@ impl SimClient {
.expect("request buffer must contain a valid request message")
}
- fn build_request(&self, operation: Operation, payload: &[u8]) -> Message<RequestHeader> {
- let header_size = std::mem::size_of::<RequestHeader>();
+ fn build_request(&self, operation: Operation, payload: &[u8]) -> Message<RoutedRequestHeader> {
+ let header_size = std::mem::size_of::<RoutedRequestHeader>();
let total_size = header_size + payload.len();
// Every `build_request` caller is a metadata-plane op (partition
// ops go through `build_request_with_namespace`), and metadata
- // requests carry the metadata consensus namespace on the wire.
- let header = self.header(operation, METADATA_CONSENSUS_NAMESPACE, total_size);
+ // requests carry the metadata consensus group on the wire.
+ let header = self.header(operation, METADATA_GROUP, total_size);
let header_bytes = bytemuck::bytes_of(&header);
let mut buffer = Vec::with_capacity(total_size);
@@ -738,8 +742,8 @@ impl SimClient {
}
#[allow(clippy::cast_possible_truncation)]
- fn header(&self, operation: Operation, namespace: u64, total_size: usize) -> RequestHeader {
- RequestHeader {
+ fn header(&self, operation: Operation, group: u64, total_size: usize) -> RoutedRequestHeader {
+ RoutedRequestHeader {
command: iggy_binary_protocol::Command2::Request,
operation,
size: total_size as u32,
@@ -755,7 +759,7 @@ impl SimClient {
timestamp: 0, // TODO: Use actual timestamp
session: self.session_id(),
request: self.request_id_for(operation),
- namespace,
+ group,
..Default::default()
}
}
@@ -772,11 +776,11 @@ const fn namespace_consumer(kind: u8, consumer_id: u32) -> WireConsumer {
}
}
-/// Decompose a namespace into the `(stream_id, topic_id, partition_id)` wire
+/// Decompose a group into the `(stream_id, topic_id, partition_id)` wire
/// identifiers the consumer-offset requests carry. Namespace ids are small
/// test values that always fit `u32`.
fn namespace_ids(ns: IggyNamespace) -> (WireIdentifier, WireIdentifier, Option<u32>) {
- let to_u32 = |v: usize| u32::try_from(v).expect("namespace id fits u32");
+ let to_u32 = |v: usize| u32::try_from(v).expect("group id fits u32");
(
WireIdentifier::Numeric(to_u32(ns.stream_id())),
WireIdentifier::Numeric(to_u32(ns.topic_id())),
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 90094b893..be1ff4835 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -1610,13 +1610,16 @@ mod tests {
// shifts the trace. Re-lock on intentional changes; expect re-locks until
// error discriminants and reply bodies stabilize the wire format.
//
- // Re-locked when the sim adopted METADATA_CONSENSUS_NAMESPACE (1<<63)
+ // Re-locked when the sim adopted METADATA_GROUP (1<<63)
// for metadata requests and the metadata consensus group, replacing
// the sim-only 0: reply headers and the per-group timeout-jitter seed
// (replica_id ^ namespace) both changed. The old 0 only ever routed
// correctly because `hash % 1 == 0` at one shard per replica.
+ // Re-locked again when replies stopped echoing a group id (the
+ // client wire lost its namespace field): the reply-hash tuple
+ // dropped that component.
assert_eq!(
- h1, 0x530D_499C_5DBE_A2BE,
+ h1, 0xCF1F_BC79_B44A_65F7,
"workload reply hash drifted from locked baseline"
);
}
@@ -2262,15 +2265,7 @@ mod tests {
}
for reply in sim.step() {
let h = reply.header();
- (
- h.client,
- h.request,
- h.op,
- h.commit,
- h.namespace,
- h.operation as u8,
- )
- .hash(&mut hasher);
+ (h.client, h.request, h.op, h.commit, h.operation as u8).hash(&mut hasher);
let cmds = wl.on_reply(&reply);
apply_sim_commands(&mut sim, &cmds);
replies_seen += 1;
@@ -2894,7 +2889,7 @@ mod tests {
let header: &PrepareOkHeader = bytemuck::checked::from_bytes(
&packet.message.as_slice()[..std::mem::size_of::<PrepareOkHeader>()],
);
- header.namespace == BLOCKED_NS.load(Ordering::Relaxed)
+ header.group == BLOCKED_NS.load(Ordering::Relaxed)
}
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
@@ -2950,13 +2945,16 @@ mod tests {
// ns_b shares the client, the replicas, and the shard, but has its
// own consensus group: it must commit while ns_a stays wedged.
let msg = client.send_messages(ns_b, &[Bytes::from_static(b"independent")]);
+ // Replies no longer carry a group id; correlate by the request id
+ // this send was stamped with.
+ let ns_b_request = msg.header().request;
sim.submit_request(client_id, 0, msg.into_generic());
let mut independent_replies = 0usize;
for _ in 0..100 {
for reply in sim.step() {
assert_eq!(
- reply.header().namespace,
- ns_b.inner(),
+ reply.header().request,
+ ns_b_request,
"only ns_b may commit while ns_a's acks are blocked"
);
independent_replies += 1;
@@ -2975,7 +2973,9 @@ mod tests {
let mut drained_replies = 0usize;
for _ in 0..800 {
for reply in sim.step() {
- if reply.header().namespace == ns_a.inner() {
+ // Only ns_a replies are outstanding once ns_b committed
+ // above, so every reply counts toward the drain.
+ if reply.header().request != ns_b_request {
drained_replies += 1;
}
}
diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs
index 9abf73c32..81ab465f0 100644
--- a/core/simulator/src/replica.rs
+++ b/core/simulator/src/replica.rs
@@ -29,7 +29,7 @@ use metadata::stm::user::{Users, UsersInner};
use metadata::{IggyMetadata, apply_committed_prepare};
use partitions::{IggyPartitions, PartitionsConfig};
use server_common::crypto;
-use server_common::sharding::{METADATA_CONSENSUS_NAMESPACE, ShardId};
+use server_common::sharding::{METADATA_GROUP, ShardId};
use server_ng::bootstrap::{ShellHandlers, ShellShardHandle, wire_shell_handlers};
use shard::shards_table::PapayaShardsTable;
use std::cell::RefCell;
@@ -182,7 +182,7 @@ pub fn new_shard(
CLUSTER_ID,
replica_id,
replica_count,
- METADATA_CONSENSUS_NAMESPACE,
+ METADATA_GROUP,
SharedSimOutbox(Rc::clone(bus)),
LocalPipeline::new(),
clock.clone(),
diff --git a/core/simulator/src/workload/auditor.rs b/core/simulator/src/workload/auditor.rs
index 2cfe2c28d..5e2bef7ee 100644
--- a/core/simulator/src/workload/auditor.rs
+++ b/core/simulator/src/workload/auditor.rs
@@ -148,16 +148,11 @@ impl ServerAuditor {
return OnReply::Unknown;
};
- // Reply's namespace must match the namespace the request was
- // submitted to. A mismatch means the reply landed in the wrong
- // VSR group's bookkeeping; refuse to apply effects against the
- // wrong shadow bucket. Entry already consumed.
- if entry.request_namespace != header.namespace {
- self.stats.replies_unknown += 1;
- return OnReply::NsMismatch;
- }
-
- let ns_key = (header.client, header.namespace);
+ // Replies no longer echo a group id (the client wire has no
+ // namespace field at all), so correlation rests entirely on the
+ // (client, request) key that fetched `entry`; the group the request
+ // was submitted to comes from the sim's own bookkeeping.
+ let ns_key = (header.client, entry.request_namespace);
let last_commit = self
.last_commit_watermark_per_client_ns
.entry(ns_key)
diff --git a/core/simulator/src/workload/mod.rs b/core/simulator/src/workload/mod.rs
index 045e946fa..2e641066c 100644
--- a/core/simulator/src/workload/mod.rs
+++ b/core/simulator/src/workload/mod.rs
@@ -40,7 +40,7 @@ use crate::workload::ops::InFlight;
use actions::Action;
use auditor::{OnReply, ServerAuditor};
use effect::SimCommand;
-use iggy_binary_protocol::{ReplyHeader, RequestHeader, result_code};
+use iggy_binary_protocol::{ReplyHeader, RoutedRequestHeader, result_code};
use invariants::Invariants;
use metadata::stm::result::result_code_recognized;
use options::WorkloadOptions;
@@ -138,7 +138,10 @@ impl Workload {
/// PRNG before `sample` runs, so they advance the trace even when `sample`
/// returns `None` (a targeted outcome whose precondition is unmet, e.g. a
/// duplicate-name target with an empty shadow). `samples_none` counts these.
- pub fn build_request(&mut self, client: &SimClient) -> Option<(u8, Message<RequestHeader>)> {
+ pub fn build_request(
+ &mut self,
+ client: &SimClient,
+ ) -> Option<(u8, Message<RoutedRequestHeader>)> {
if !self.client_idle(client.client_id()) {
return None;
}
@@ -167,7 +170,7 @@ impl Workload {
action,
input,
outcome,
- request_namespace: header.namespace,
+ request_namespace: header.group,
},
);
*self
diff --git a/core/simulator/src/workload/ops/change_password.rs b/core/simulator/src/workload/ops/change_password.rs
index edd2e0cf9..59668c92c 100644
--- a/core/simulator/src/workload/ops/change_password.rs
+++ b/core/simulator/src/workload/ops/change_password.rs
@@ -18,7 +18,7 @@
//! `ChangePassword` op. Targets `Ok` (rotate live user's password) or
//! `UserNotFound` (fabricated user).
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand::RngExt;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -68,7 +68,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.change_password(&input.user, &input.current_password, &input.new_password)
}
diff --git a/core/simulator/src/workload/ops/create_consumer_group.rs b/core/simulator/src/workload/ops/create_consumer_group.rs
index 66113e65b..df1bc5c63 100644
--- a/core/simulator/src/workload/ops/create_consumer_group.rs
+++ b/core/simulator/src/workload/ops/create_consumer_group.rs
@@ -21,7 +21,7 @@
//! (a fabricated parent stream), `TopicNotFound` (a live stream with a
//! fabricated topic), or `NameAlreadyExists` (an existing group name).
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -87,7 +87,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.create_consumer_group(&input.stream, &input.topic, &input.name)
}
diff --git a/core/simulator/src/workload/ops/create_partitions.rs b/core/simulator/src/workload/ops/create_partitions.rs
index 4ee8eb300..8b38b432a 100644
--- a/core/simulator/src/workload/ops/create_partitions.rs
+++ b/core/simulator/src/workload/ops/create_partitions.rs
@@ -22,7 +22,7 @@
//! not targeted. Shadow tracks no partition counts, so every outcome predicts
//! `Effect::None`.
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand::RngExt;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -86,7 +86,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.create_partitions(&input.stream, &input.topic, input.partitions_count)
}
diff --git a/core/simulator/src/workload/ops/create_personal_access_token.rs b/core/simulator/src/workload/ops/create_personal_access_token.rs
index da01cbc48..561b51534 100644
--- a/core/simulator/src/workload/ops/create_personal_access_token.rs
+++ b/core/simulator/src/workload/ops/create_personal_access_token.rs
@@ -19,7 +19,7 @@
//! (a live token name). `InvalidExpiry` is not targeted; tokens never expire
//! (expiry = 0).
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -56,7 +56,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.create_personal_access_token(&input.name, input.expiry)
}
diff --git a/core/simulator/src/workload/ops/create_stream.rs b/core/simulator/src/workload/ops/create_stream.rs
index c0b3047ef..cf5e68ace 100644
--- a/core/simulator/src/workload/ops/create_stream.rs
+++ b/core/simulator/src/workload/ops/create_stream.rs
@@ -18,7 +18,7 @@
//! `CreateStream` op. Targets `Ok` with a fresh name, or `NameAlreadyExists`
//! by reusing a live stream name from the shadow.
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -53,7 +53,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.create_stream(&input.name)
}
diff --git a/core/simulator/src/workload/ops/create_topic.rs b/core/simulator/src/workload/ops/create_topic.rs
index 7d7b2f761..07fd654a1 100644
--- a/core/simulator/src/workload/ops/create_topic.rs
+++ b/core/simulator/src/workload/ops/create_topic.rs
@@ -19,7 +19,7 @@
//! `StreamNotFound` (a fabricated parent stream), or `NameAlreadyExists` (an
//! existing topic name under its live stream).
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand::RngExt;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -84,7 +84,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.create_topic(&input.stream, &input.name, input.partitions_count)
}
diff --git a/core/simulator/src/workload/ops/create_user.rs b/core/simulator/src/workload/ops/create_user.rs
index 55fb8993e..c5bce99db 100644
--- a/core/simulator/src/workload/ops/create_user.rs
+++ b/core/simulator/src/workload/ops/create_user.rs
@@ -18,7 +18,7 @@
//! `CreateUser` op. Targets `Ok` (fresh username) or `UserAlreadyExists` (a
//! live username from the shadow). Status fixed at 1 (Active).
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -59,7 +59,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.create_user(&input.username, &input.password, input.status)
}
diff --git a/core/simulator/src/workload/ops/delete_consumer_group.rs b/core/simulator/src/workload/ops/delete_consumer_group.rs
index cfbe89961..996ea1d3c 100644
--- a/core/simulator/src/workload/ops/delete_consumer_group.rs
+++ b/core/simulator/src/workload/ops/delete_consumer_group.rs
@@ -22,7 +22,7 @@
//! `ConsumerGroupNotFound` (a live stream/topic with a fabricated group
//! name), mirroring the legacy resolution ladder.
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -86,7 +86,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.delete_consumer_group(&input.stream, &input.topic, &input.group)
}
diff --git a/core/simulator/src/workload/ops/delete_consumer_offset.rs b/core/simulator/src/workload/ops/delete_consumer_offset.rs
index 0c71b6031..c05c7960a 100644
--- a/core/simulator/src/workload/ops/delete_consumer_offset.rs
+++ b/core/simulator/src/workload/ops/delete_consumer_offset.rs
@@ -17,7 +17,7 @@
//! `DeleteConsumerOffset` op. Live namespace via shadow.
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand::RngExt;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -63,7 +63,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.delete_consumer_offset(input.ns, input.consumer_kind, input.consumer_id)
}
diff --git a/core/simulator/src/workload/ops/delete_consumer_offset_2.rs b/core/simulator/src/workload/ops/delete_consumer_offset_2.rs
index 5913b018d..ad2a6201a 100644
--- a/core/simulator/src/workload/ops/delete_consumer_offset_2.rs
+++ b/core/simulator/src/workload/ops/delete_consumer_offset_2.rs
@@ -17,7 +17,7 @@
//! `DeleteConsumerOffset2` op. Namespace-routed with `AckLevel`.
-use iggy_binary_protocol::{AckLevel, RequestHeader};
+use iggy_binary_protocol::{AckLevel, RoutedRequestHeader};
use rand::RngExt;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -71,7 +71,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.delete_consumer_offset_2(input.ns, input.consumer_kind, input.consumer_id, input.ack)
}
diff --git a/core/simulator/src/workload/ops/delete_partitions.rs b/core/simulator/src/workload/ops/delete_partitions.rs
index 46ee23e73..3b08bc797 100644
--- a/core/simulator/src/workload/ops/delete_partitions.rs
+++ b/core/simulator/src/workload/ops/delete_partitions.rs
@@ -21,7 +21,7 @@
//! or `TopicNotFound` (a live stream with a fabricated topic). Partition counts
//! are not tracked in the shadow, so every outcome predicts `Effect::None`.
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand::RngExt;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -82,7 +82,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.delete_partitions(&input.stream, &input.topic, input.partitions_count)
}
diff --git a/core/simulator/src/workload/ops/delete_personal_access_token.rs b/core/simulator/src/workload/ops/delete_personal_access_token.rs
index 2bd54304e..a829f5fa7 100644
--- a/core/simulator/src/workload/ops/delete_personal_access_token.rs
+++ b/core/simulator/src/workload/ops/delete_personal_access_token.rs
@@ -18,7 +18,7 @@
//! `DeletePersonalAccessToken` op. Targets `Ok` (a live token) or `NotFound`
//! (a fabricated name).
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -51,7 +51,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.delete_personal_access_token(&input.name)
}
diff --git a/core/simulator/src/workload/ops/delete_segments.rs b/core/simulator/src/workload/ops/delete_segments.rs
index 399016f77..abbb173ff 100644
--- a/core/simulator/src/workload/ops/delete_segments.rs
+++ b/core/simulator/src/workload/ops/delete_segments.rs
@@ -26,7 +26,7 @@
//! in `Action` and the dispatch table so the surface compiles and the v2.4
//! outcome expansion lands cleanly post-upgrade.
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -61,7 +61,7 @@ pub const fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.delete_segments(
&input.stream,
&input.topic,
diff --git a/core/simulator/src/workload/ops/delete_stream.rs b/core/simulator/src/workload/ops/delete_stream.rs
index 008b9f04a..97aa7f050 100644
--- a/core/simulator/src/workload/ops/delete_stream.rs
+++ b/core/simulator/src/workload/ops/delete_stream.rs
@@ -18,7 +18,7 @@
//! `DeleteStream` op. Targets `Ok` with a live stream name, or `StreamNotFound`
//! with a fabricated name that was never created.
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -51,7 +51,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.delete_stream(&input.name)
}
diff --git a/core/simulator/src/workload/ops/delete_topic.rs b/core/simulator/src/workload/ops/delete_topic.rs
index 32e640ba7..4dfd8241f 100644
--- a/core/simulator/src/workload/ops/delete_topic.rs
+++ b/core/simulator/src/workload/ops/delete_topic.rs
@@ -19,7 +19,7 @@
//! fabricated parent stream), or `TopicNotFound` (a live stream with a
//! fabricated topic).
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -63,7 +63,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.delete_topic(&input.stream, &input.topic)
}
diff --git a/core/simulator/src/workload/ops/delete_user.rs b/core/simulator/src/workload/ops/delete_user.rs
index 964b8eee8..55c039eb7 100644
--- a/core/simulator/src/workload/ops/delete_user.rs
+++ b/core/simulator/src/workload/ops/delete_user.rs
@@ -18,7 +18,7 @@
//! `DeleteUser` op. Targets `Ok` (a live user) or `UserNotFound` (a fabricated
//! username).
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -53,7 +53,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.delete_user(&input.user)
}
diff --git a/core/simulator/src/workload/ops/mod.rs b/core/simulator/src/workload/ops/mod.rs
index be965e910..815976072 100644
--- a/core/simulator/src/workload/ops/mod.rs
+++ b/core/simulator/src/workload/ops/mod.rs
@@ -55,7 +55,7 @@ pub mod update_stream;
pub mod update_topic;
pub mod update_user;
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -130,7 +130,7 @@ macro_rules! op_dispatch {
}
#[must_use]
- pub fn build_message(client: &SimClient, input: &InFlightInput) -> Message<RequestHeader> {
+ pub fn build_message(client: &SimClient, input: &InFlightInput) -> Message<RoutedRequestHeader> {
match input {
$( InFlightInput::$variant(i) => $module::build_message(client, i), )*
}
diff --git a/core/simulator/src/workload/ops/purge_stream.rs b/core/simulator/src/workload/ops/purge_stream.rs
index 50033df79..a7f89c66d 100644
--- a/core/simulator/src/workload/ops/purge_stream.rs
+++ b/core/simulator/src/workload/ops/purge_stream.rs
@@ -18,7 +18,7 @@
//! `PurgeStream` op. Targets `Ok` (live stream) or `StreamNotFound`
//! (fabricated, never-created name).
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -51,7 +51,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.purge_stream(&input.stream)
}
diff --git a/core/simulator/src/workload/ops/purge_topic.rs b/core/simulator/src/workload/ops/purge_topic.rs
index 00ae96e30..875c62544 100644
--- a/core/simulator/src/workload/ops/purge_topic.rs
+++ b/core/simulator/src/workload/ops/purge_topic.rs
@@ -18,7 +18,7 @@
//! `PurgeTopic` op. Targets `Ok` (live topic), `StreamNotFound` (fabricated
//! parent stream), or `TopicNotFound` (live stream, fabricated topic).
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -62,7 +62,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.purge_topic(&input.stream, &input.topic)
}
diff --git a/core/simulator/src/workload/ops/send_messages.rs b/core/simulator/src/workload/ops/send_messages.rs
index 7e58a27f9..fda0c0656 100644
--- a/core/simulator/src/workload/ops/send_messages.rs
+++ b/core/simulator/src/workload/ops/send_messages.rs
@@ -22,7 +22,7 @@
//! 3. one `prng.random()` per payload to disambiguate body bytes
use bytes::Bytes;
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand::RngExt;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -74,7 +74,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.send_messages(input.ns, &input.payloads)
}
diff --git a/core/simulator/src/workload/ops/store_consumer_offset.rs b/core/simulator/src/workload/ops/store_consumer_offset.rs
index feda1b47c..4c26a0b82 100644
--- a/core/simulator/src/workload/ops/store_consumer_offset.rs
+++ b/core/simulator/src/workload/ops/store_consumer_offset.rs
@@ -18,7 +18,7 @@
//! `StoreConsumerOffset` op. Pre-`AckLevel` manual encoding. Live
//! namespace via shadow, fabricated consumer kind/id. Samples Success.
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand::RngExt;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -73,7 +73,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.store_consumer_offset(
input.ns,
input.consumer_kind,
diff --git a/core/simulator/src/workload/ops/store_consumer_offset_2.rs b/core/simulator/src/workload/ops/store_consumer_offset_2.rs
index c4b2785a8..083f7a887 100644
--- a/core/simulator/src/workload/ops/store_consumer_offset_2.rs
+++ b/core/simulator/src/workload/ops/store_consumer_offset_2.rs
@@ -23,7 +23,7 @@
//! 4. `offset` range draw
//! 5. `ack` ratio draw
-use iggy_binary_protocol::{AckLevel, RequestHeader};
+use iggy_binary_protocol::{AckLevel, RoutedRequestHeader};
use rand::RngExt;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -86,7 +86,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.store_consumer_offset_2(
input.ns,
input.consumer_kind,
diff --git a/core/simulator/src/workload/ops/update_permissions.rs b/core/simulator/src/workload/ops/update_permissions.rs
index 75169a382..b3aad70e8 100644
--- a/core/simulator/src/workload/ops/update_permissions.rs
+++ b/core/simulator/src/workload/ops/update_permissions.rs
@@ -18,7 +18,7 @@
//! `UpdatePermissions` op. Targets `Ok` (live user) or `UserNotFound`
//! (fabricated user). No permissions payload, so every outcome is `Effect::None`.
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -53,7 +53,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.update_permissions(&input.user)
}
diff --git a/core/simulator/src/workload/ops/update_stream.rs b/core/simulator/src/workload/ops/update_stream.rs
index ef70236d8..f6f228509 100644
--- a/core/simulator/src/workload/ops/update_stream.rs
+++ b/core/simulator/src/workload/ops/update_stream.rs
@@ -21,7 +21,7 @@
//! (fabricated stream). `NameAlreadyExists` (rename onto a live name) not
//! targeted, but the server still classifies it on a race.
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -63,7 +63,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.update_stream(&input.stream, &input.new_name)
}
diff --git a/core/simulator/src/workload/ops/update_topic.rs b/core/simulator/src/workload/ops/update_topic.rs
index a06f2a61d..a4e36367e 100644
--- a/core/simulator/src/workload/ops/update_topic.rs
+++ b/core/simulator/src/workload/ops/update_topic.rs
@@ -21,7 +21,7 @@
//! parent stream), or `TopicNotFound` (live stream, fabricated topic).
//! `NameAlreadyExists` not targeted.
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -77,7 +77,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.update_topic(&input.stream, &input.topic, &input.new_name)
}
diff --git a/core/simulator/src/workload/ops/update_user.rs b/core/simulator/src/workload/ops/update_user.rs
index 045c377b1..c0c15aade 100644
--- a/core/simulator/src/workload/ops/update_user.rs
+++ b/core/simulator/src/workload/ops/update_user.rs
@@ -18,7 +18,7 @@
//! `UpdateUser` op. Targets `Ok` (rename live user to fresh name) or
//! `UserNotFound` (fabricated user). `UsernameAlreadyExists` not targeted.
-use iggy_binary_protocol::RequestHeader;
+use iggy_binary_protocol::RoutedRequestHeader;
use rand_xoshiro::Xoshiro256Plus;
use server_common::Message;
@@ -74,7 +74,7 @@ pub fn sample(
}
#[must_use]
-pub fn build_message(client: &SimClient, input: &Input) -> Message<RequestHeader> {
+pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> {
client.update_user(&input.user, input.new_username.as_deref(), input.status)
}