This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch feat/server-ng-perf in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 1b8f3dbd5acbd5d484e3f43c1a2d6e578e515f0c Author: Hubert Gruszecki <[email protected]> AuthorDate: Fri Jul 24 16:30:16 2026 +0200 perf(server-ng): reach single-node latency parity with legacy server server-ng trailed the legacy server on single-node benchmarks: the default cpu allocation collapsed all work onto one shard, produce hashed each batch three times (convert, stamp, flush revalidation), and a lone consumer crossing a sealed segment degraded to full-segment scans through unbounded read handles. Shard allocation now defaults to numa:auto. Produce computes the batch checksum once and marks locally originated batches trusted; replicated blobs keep a per-message receive gate on followers, so transit integrity still holds end to end. Sealed segment read handles are capped by a per-partition LRU. Consumer groups on the ng metadata plane gain legacy-parity error semantics: join/leave against a missing stream, topic or group fail loudly (1009/2010/5000), leaving as an absent member returns 5006 instead of silent success, and the wire codes are pinned by an integration test across TCP, QUIC and WebSocket. The HTTP listener now binds from the cluster roster port, so one-host clusters run from byte-identical per-node configs. --- .../tests/sdk/consumer_group_membership.rs | 98 ++++ core/journal/src/prepare_journal.rs | 9 +- core/metadata/src/stm/consumer_group.rs | 207 ++++++- core/metadata/src/stm/result.rs | 52 +- core/metadata/src/stm/snapshot.rs | 2 +- core/metadata/src/stm/stream.rs | 25 +- core/partitions/src/iggy_index_reader.rs | 35 +- core/partitions/src/iggy_partition.rs | 263 ++++++++- core/partitions/src/iggy_partitions.rs | 23 +- core/partitions/src/journal.rs | 10 +- core/partitions/src/log.rs | 177 ++++++ core/partitions/src/poll_plan.rs | 163 +++++- core/server-ng/config.toml | 3 +- core/server-ng/src/bootstrap.rs | 120 +++- core/server-ng/src/dispatch.rs | 8 +- core/server-ng/src/partition_reconciler.rs | 24 +- core/server-ng/src/responses.rs | 2 +- core/server_common/src/send_messages2.rs | 626 ++++++++++++++++++--- core/simulator/src/client.rs | 4 +- 19 files changed, 1665 insertions(+), 186 deletions(-) diff --git a/core/integration/tests/sdk/consumer_group_membership.rs b/core/integration/tests/sdk/consumer_group_membership.rs index 76d413a48..f7c36aa3e 100644 --- a/core/integration/tests/sdk/consumer_group_membership.rs +++ b/core/integration/tests/sdk/consumer_group_membership.rs @@ -32,6 +32,14 @@ const PARK_TIMEOUT: Duration = Duration::from_secs(2); // Generous bound: on regression the poll hangs until this elapses. const RESOLVE_TIMEOUT: Duration = Duration::from_secs(10); +// Legacy wire error codes for the join/leave failure ladder. Pinned as literals +// (not derived from `IggyError`) so renumbering the wire contract fails here +// loudly -- the same guarantee the non-Rust SDKs depend on. +const STREAM_ID_NOT_FOUND: u32 = 1009; +const TOPIC_ID_NOT_FOUND: u32 = 2010; +const CONSUMER_GROUP_ID_NOT_FOUND: u32 = 5000; +const CONSUMER_GROUP_MEMBER_NOT_FOUND: u32 = 5006; + // A consumer-group member holding zero partitions has the same empty client-side // assignment as a non-member; only membership tells them apart. When the group // is deleted under such a member, the poll must surface an error (driving a @@ -137,3 +145,93 @@ async fn given_group_member_holds_no_partitions_when_group_deleted_should_surfac Ok(_) => panic!("expected an error after group deletion, got a message"), } } + +// End-to-end wire pin for the consumer-group join/leave error ladder. The +// metadata STM unit tests pin the committed result codes; this pins that +// server-ng actually emits them over the wire, so a client observes the same +// codes the legacy server returns. The full ladder runs per binary transport, +// since the error response is encoded per transport. +#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic])] +async fn given_join_and_leave_failures_when_sent_over_the_wire_should_return_legacy_error_codes( + harness: &TestHarness, +) { + let root_client = harness.root_client().await.expect("root client"); + let stream_id = Identifier::named(STREAM_NAME).unwrap(); + let topic_id = Identifier::named(TOPIC_NAME).unwrap(); + let group_id = Identifier::named(CONSUMER_GROUP_NAME).unwrap(); + // Stands in for whichever level is absent; its position in the call marks + // the level under test. + let missing = Identifier::named("cg-membership-nonexistent").unwrap(); + + // Each level is created only after the case proving its absence is + // rejected, so no assertion resolves against pre-existing state. + + // Nothing exists yet: the stream miss is the first rejection. + assert_rejected( + root_client + .join_consumer_group(&missing, &topic_id, &group_id) + .await, + STREAM_ID_NOT_FOUND, + "join with a missing stream", + ); + + root_client.create_stream(STREAM_NAME).await.unwrap(); + assert_rejected( + root_client + .join_consumer_group(&stream_id, &missing, &group_id) + .await, + TOPIC_ID_NOT_FOUND, + "join with a missing topic", + ); + + root_client + .create_topic( + &stream_id, + TOPIC_NAME, + 1, + CompressionAlgorithm::default(), + None, + IggyExpiry::NeverExpire, + MaxTopicSize::ServerDefault, + ) + .await + .unwrap(); + assert_rejected( + root_client + .join_consumer_group(&stream_id, &topic_id, &missing) + .await, + CONSUMER_GROUP_ID_NOT_FOUND, + "join with a missing group", + ); + + root_client + .create_consumer_group(&stream_id, &topic_id, CONSUMER_GROUP_NAME) + .await + .unwrap(); + // The group resolves now, but this client never joined it: the member check + // is the loud rejection, distinct from the missing-group case above. + assert_rejected( + root_client + .leave_consumer_group(&stream_id, &topic_id, &group_id) + .await, + CONSUMER_GROUP_MEMBER_NOT_FOUND, + "leave a group the client never joined", + ); + + // Non-vacuous control: the same client joins the group whose leave just + // returned member-not-found, proving the setup is live. + root_client + .join_consumer_group(&stream_id, &topic_id, &group_id) + .await + .expect("join on an existing group must succeed"); +} + +fn assert_rejected(result: Result<(), IggyError>, expected_code: u32, context: &str) { + let error = result.expect_err(context); + assert_eq!( + error.as_code(), + expected_code, + "{context}: expected wire error code {expected_code}, got {} ({error})", + error.as_code(), + ); +} diff --git a/core/journal/src/prepare_journal.rs b/core/journal/src/prepare_journal.rs index 802b36d81..6acb94726 100644 --- a/core/journal/src/prepare_journal.rs +++ b/core/journal/src/prepare_journal.rs @@ -350,9 +350,12 @@ impl PrepareJournal { // `PrepareHeader` projection in consensus builds prepares with // `..Default::default()` so the integrity fields are always 0. // Until a producer computes them, verification here would be - // trivially-passing noise. Without it, a body bit-flip that - // leaves the header valid is replayed silently as corrupt - // state. Committed bytes are meant to be byte-identical across + // trivially-passing noise. The per-message receive gate now rejects + // transit body corruption on a follower before apply, but this + // recovery scan still does not re-verify body integrity read back + // from disk, so an at-rest body bit-flip that leaves the header + // valid is replayed silently. Committed bytes are meant to be + // byte-identical across // replicas (deterministic apply, timestamp replicated not // re-projected), so once the producer computes the integrity fields // they should agree on every node and this check can be turned on diff --git a/core/metadata/src/stm/consumer_group.rs b/core/metadata/src/stm/consumer_group.rs index c85390ad9..340eb4b85 100644 --- a/core/metadata/src/stm/consumer_group.rs +++ b/core/metadata/src/stm/consumer_group.rs @@ -27,7 +27,10 @@ //! offset -- making a metadata->offset purge unnecessary for correctness. use crate::stm::StateHandler; -use crate::stm::result::{ApplyReply, CreateConsumerGroupResult, DeleteConsumerGroupResult}; +use crate::stm::result::{ + ApplyReply, CreateConsumerGroupResult, DeleteConsumerGroupResult, JoinConsumerGroupResult, + LeaveConsumerGroupResult, +}; use crate::stm::stream::StreamsInner; use bytes::Bytes; @@ -589,17 +592,30 @@ impl StateHandler for DeleteConsumerGroupRequest { impl StateHandler for JoinConsumerGroupRequest { type State = StreamsInner; fn apply(&self, state: &mut StreamsInner, timestamp: iggy_common::IggyTimestamp) -> ApplyReply { - let Some(topic) = state.topic_mut(&self.stream_id, &self.topic_id) else { - return ApplyReply::ok(Bytes::new()); + // Resolve level by level so the committed rejection names the level that + // missed, mirroring the legacy `resolve_consumer_group` ladder instead of + // the old silent-OK no-op. + let Some(stream_id) = state.resolve_stream_id(&self.stream_id) else { + return ApplyReply::err(JoinConsumerGroupResult::StreamNotFound); + }; + let Some(topic_id) = state.resolve_topic_id(stream_id, &self.topic_id) else { + return ApplyReply::err(JoinConsumerGroupResult::TopicNotFound); + }; + let Some(topic) = state + .items + .get_mut(stream_id) + .and_then(|stream| stream.topics.get_mut(topic_id)) + else { + return ApplyReply::err(JoinConsumerGroupResult::TopicNotFound); }; let Some(group_id) = topic.resolve_group_id(&self.group_id) else { - return ApplyReply::ok(Bytes::new()); + return ApplyReply::err(JoinConsumerGroupResult::ConsumerGroupNotFound); }; // Snapshot the live partition ids before taking a mutable borrow of the // group (both borrow the topic). let partition_ids: Vec<usize> = topic.partitions.iter().map(|p| p.id).collect(); let Some(group) = topic.consumer_groups.get_mut(&group_id) else { - return ApplyReply::ok(Bytes::new()); + return ApplyReply::err(JoinConsumerGroupResult::ConsumerGroupNotFound); }; // Idempotent: a re-join from the same client keeps its membership. let already = group @@ -633,26 +649,43 @@ impl StateHandler for LeaveConsumerGroupRequest { state: &mut StreamsInner, _timestamp: iggy_common::IggyTimestamp, ) -> ApplyReply { - let Some(topic) = state.topic_mut(&self.stream_id, &self.topic_id) else { - return ApplyReply::ok(Bytes::new()); + // Same level-by-level resolution as Join, surfacing a loud rejection + // instead of the old silent-OK no-op when a level is gone. + let Some(stream_id) = state.resolve_stream_id(&self.stream_id) else { + return ApplyReply::err(LeaveConsumerGroupResult::StreamNotFound); + }; + let Some(topic_id) = state.resolve_topic_id(stream_id, &self.topic_id) else { + return ApplyReply::err(LeaveConsumerGroupResult::TopicNotFound); + }; + let Some(topic) = state + .items + .get_mut(stream_id) + .and_then(|stream| stream.topics.get_mut(topic_id)) + else { + return ApplyReply::err(LeaveConsumerGroupResult::TopicNotFound); }; let Some(group_id) = topic.resolve_group_id(&self.group_id) else { - return ApplyReply::ok(Bytes::new()); + return ApplyReply::err(LeaveConsumerGroupResult::ConsumerGroupNotFound); }; let partition_ids: Vec<usize> = topic.partitions.iter().map(|p| p.id).collect(); let Some(group) = topic.consumer_groups.get_mut(&group_id) else { - return ApplyReply::ok(Bytes::new()); + return ApplyReply::err(LeaveConsumerGroupResult::ConsumerGroupNotFound); }; let member_key = group .members .iter() .find(|(_, m)| m.client_id == self.client_id) .map(|(key, _)| key); - if let Some(key) = member_key { - group.members.remove(key); - group.rebalance_members(&partition_ids); - state.recompute_pending_revocations_count(); - } + let Some(key) = member_key else { + // Group exists but this client never joined it: mirror legacy's + // ConsumerGroupMemberNotFound instead of a silent no-op. The + // server-internal RemoveConsumerGroupMember (disconnect cleanup) + // stays idempotent -- only this client-facing Leave is loud. + return ApplyReply::err(LeaveConsumerGroupResult::MemberNotFound); + }; + group.members.remove(key); + group.rebalance_members(&partition_ids); + state.recompute_pending_revocations_count(); ApplyReply::ok(Bytes::new()) } } @@ -958,6 +991,45 @@ mod tests { ) } + fn join( + state: &mut StreamsInner, + stream: u32, + topic: u32, + group: u32, + client_id: u128, + ) -> ApplyReply { + StateHandler::apply( + &JoinConsumerGroupRequest { + stream_id: WireIdentifier::numeric(stream), + topic_id: WireIdentifier::numeric(topic), + group_id: WireIdentifier::numeric(group), + client_id, + in_flight: Vec::new(), + }, + state, + IggyTimestamp::now(), + ) + } + + fn leave( + state: &mut StreamsInner, + stream: u32, + topic: u32, + group: u32, + client_id: u128, + ) -> ApplyReply { + StateHandler::apply( + &LeaveConsumerGroupRequest { + stream_id: WireIdentifier::numeric(stream), + topic_id: WireIdentifier::numeric(topic), + group_id: WireIdentifier::numeric(group), + client_id, + }, + state, + IggyTimestamp::now(), + ) + } + #[test] fn given_duplicate_name_when_apply_create_consumer_group_should_return_name_already_exists() { let mut state = streams_with_topic(); @@ -1021,4 +1093,111 @@ mod tests { assert_eq!(apply.code, u32::from(DeleteConsumerGroupResult::NotFound)); assert!(apply.body.is_empty()); } + + #[test] + fn given_existing_group_when_apply_join_consumer_group_should_succeed() { + let mut state = streams_with_topic(); + // Groups are 0-based, so the first created group is id 0. + assert_eq!(create_group(&mut state, "group").code, 0); + + let apply = join(&mut state, 0, 0, 0, 1); + assert_eq!(apply.code, 0); + assert!(apply.body.is_empty()); + + // A re-join from the same client is an idempotent success. + assert_eq!(join(&mut state, 0, 0, 0, 1).code, 0); + } + + #[test] + fn given_missing_stream_when_apply_join_consumer_group_should_return_stream_not_found() { + let mut state = streams_with_topic(); + let apply = join(&mut state, 999, 0, 0, 1); + assert_eq!( + apply.code, + u32::from(JoinConsumerGroupResult::StreamNotFound) + ); + assert!(apply.body.is_empty()); + } + + #[test] + fn given_missing_topic_when_apply_join_consumer_group_should_return_topic_not_found() { + let mut state = streams_with_topic(); + let apply = join(&mut state, 0, 999, 0, 1); + assert_eq!( + apply.code, + u32::from(JoinConsumerGroupResult::TopicNotFound) + ); + assert!(apply.body.is_empty()); + } + + #[test] + fn given_missing_group_when_apply_join_consumer_group_should_return_consumer_group_not_found() { + let mut state = streams_with_topic(); + let apply = join(&mut state, 0, 0, 999, 1); + assert_eq!( + apply.code, + u32::from(JoinConsumerGroupResult::ConsumerGroupNotFound) + ); + assert!(apply.body.is_empty()); + } + + #[test] + fn given_joined_member_when_apply_leave_consumer_group_should_succeed() { + let mut state = streams_with_topic(); + assert_eq!(create_group(&mut state, "group").code, 0); + assert_eq!(join(&mut state, 0, 0, 0, 1).code, 0); + + let apply = leave(&mut state, 0, 0, 0, 1); + assert_eq!(apply.code, 0); + assert!(apply.body.is_empty()); + } + + #[test] + fn given_absent_member_when_apply_leave_consumer_group_should_return_member_not_found() { + let mut state = streams_with_topic(); + assert_eq!(create_group(&mut state, "group").code, 0); + assert_eq!(join(&mut state, 0, 0, 0, 1).code, 0); + + // Client 2 is not a member of the group, which does exist. + let apply = leave(&mut state, 0, 0, 0, 2); + assert_eq!( + apply.code, + u32::from(LeaveConsumerGroupResult::MemberNotFound) + ); + assert!(apply.body.is_empty()); + } + + #[test] + fn given_missing_stream_when_apply_leave_consumer_group_should_return_stream_not_found() { + let mut state = streams_with_topic(); + let apply = leave(&mut state, 999, 0, 0, 1); + assert_eq!( + apply.code, + u32::from(LeaveConsumerGroupResult::StreamNotFound) + ); + assert!(apply.body.is_empty()); + } + + #[test] + fn given_missing_topic_when_apply_leave_consumer_group_should_return_topic_not_found() { + let mut state = streams_with_topic(); + let apply = leave(&mut state, 0, 999, 0, 1); + assert_eq!( + apply.code, + u32::from(LeaveConsumerGroupResult::TopicNotFound) + ); + assert!(apply.body.is_empty()); + } + + #[test] + fn given_missing_group_when_apply_leave_consumer_group_should_return_consumer_group_not_found() + { + let mut state = streams_with_topic(); + let apply = leave(&mut state, 0, 0, 999, 1); + assert_eq!( + apply.code, + u32::from(LeaveConsumerGroupResult::ConsumerGroupNotFound) + ); + assert!(apply.body.is_empty()); + } } diff --git a/core/metadata/src/stm/result.rs b/core/metadata/src/stm/result.rs index f3be7f775..ad419d593 100644 --- a/core/metadata/src/stm/result.rs +++ b/core/metadata/src/stm/result.rs @@ -231,6 +231,23 @@ result_enum!(CreateConsumerGroupResult { NameAlreadyExists = 5004, }); result_enum!(DeleteConsumerGroupResult { NotFound = 5000 }); +// Join/Leave resolve stream -> topic -> group inside the apply (the authz gate +// passes a resolution miss through), so their codes mirror the legacy +// `resolve_consumer_group` ladder -- StreamIdNotFound / TopicIdNotFound / +// ConsumerGroupIdNotFound -- and a client sees the identical failure on either +// server. Leave also mirrors legacy's post-resolution member check: leaving a +// group the client never joined returns ConsumerGroupMemberNotFound. +result_enum!(JoinConsumerGroupResult { + StreamNotFound = 1009, + TopicNotFound = 2010, + ConsumerGroupNotFound = 5000, +}); +result_enum!(LeaveConsumerGroupResult { + StreamNotFound = 1009, + TopicNotFound = 2010, + ConsumerGroupNotFound = 5000, + MemberNotFound = 5006, +}); /// `IggyError::Unauthorized`. Any control-plane op can commit as an in-apply /// authorization no-op, so this code is valid for every op regardless of its @@ -282,6 +299,8 @@ pub const fn result_code_recognized(operation: Operation, code: u32) -> bool { } Operation::CreateConsumerGroup => CreateConsumerGroupResult::from_u32(code).is_some(), Operation::DeleteConsumerGroup => DeleteConsumerGroupResult::from_u32(code).is_some(), + Operation::JoinConsumerGroup => JoinConsumerGroupResult::from_u32(code).is_some(), + Operation::LeaveConsumerGroup => LeaveConsumerGroupResult::from_u32(code).is_some(), _ => true, } } @@ -561,9 +580,40 @@ mod tests { u32::from(CreateConsumerGroupResult::NameAlreadyExists), IggyError::ConsumerGroupNameAlreadyExists(String::new(), id()).as_code(), ); + let consumer_group_not_found = IggyError::ConsumerGroupIdNotFound(id(), id()).as_code(); assert_eq!( u32::from(DeleteConsumerGroupResult::NotFound), - IggyError::ConsumerGroupIdNotFound(id(), id()).as_code(), + consumer_group_not_found, + ); + + // Join/Leave mirror the legacy `resolve_consumer_group` error ladder. + assert_eq!( + u32::from(JoinConsumerGroupResult::StreamNotFound), + stream_not_found + ); + assert_eq!( + u32::from(JoinConsumerGroupResult::TopicNotFound), + topic_not_found + ); + assert_eq!( + u32::from(JoinConsumerGroupResult::ConsumerGroupNotFound), + consumer_group_not_found, + ); + assert_eq!( + u32::from(LeaveConsumerGroupResult::StreamNotFound), + stream_not_found + ); + assert_eq!( + u32::from(LeaveConsumerGroupResult::TopicNotFound), + topic_not_found + ); + assert_eq!( + u32::from(LeaveConsumerGroupResult::ConsumerGroupNotFound), + consumer_group_not_found, + ); + assert_eq!( + u32::from(LeaveConsumerGroupResult::MemberNotFound), + IggyError::ConsumerGroupMemberNotFound(0, id(), id()).as_code(), ); // Unauthorized (41) - the global in-apply RBAC denial code. diff --git a/core/metadata/src/stm/snapshot.rs b/core/metadata/src/stm/snapshot.rs index 8d61cb23e..ee59f14c9 100644 --- a/core/metadata/src/stm/snapshot.rs +++ b/core/metadata/src/stm/snapshot.rs @@ -361,7 +361,7 @@ mod tests { purge_generation: 0, }], consumer_groups: Vec::new(), - next_consumer_group_id: 1, + next_consumer_group_id: 0, }, )], }, diff --git a/core/metadata/src/stm/stream.rs b/core/metadata/src/stm/stream.rs index 1b5f6b9e6..bde39d335 100644 --- a/core/metadata/src/stm/stream.rs +++ b/core/metadata/src/stm/stream.rs @@ -201,7 +201,7 @@ impl Default for Topic { round_robin_counter: Arc::new(AtomicUsize::new(0)), consumer_groups: AHashMap::default(), consumer_group_index: AHashMap::default(), - next_consumer_group_id: 1, + next_consumer_group_id: 0, } } } @@ -229,7 +229,7 @@ impl Topic { round_robin_counter: Arc::new(AtomicUsize::new(0)), consumer_groups: AHashMap::default(), consumer_group_index: AHashMap::default(), - next_consumer_group_id: 1, + next_consumer_group_id: 0, } } @@ -1416,7 +1416,7 @@ impl StateHandler for CreateTopicWithAssignmentsRequest { round_robin_counter: Arc::new(AtomicUsize::new(0)), consumer_groups: AHashMap::default(), consumer_group_index: AHashMap::default(), - next_consumer_group_id: 1, + next_consumer_group_id: 0, }; let inserted = stream.topics.insert(topic); @@ -1862,17 +1862,14 @@ impl Snapshotable for Streams { .iter() .map(|(_, group_snap)| (Arc::from(group_snap.name.as_str()), group_snap.id)) .collect(), - next_consumer_group_id: topic_snap - .next_consumer_group_id - .max( - topic_snap - .consumer_groups - .iter() - .map(|(id, _)| id + 1) - .max() - .unwrap_or(1), - ) - .max(1), + next_consumer_group_id: topic_snap.next_consumer_group_id.max( + topic_snap + .consumer_groups + .iter() + .map(|(id, _)| id + 1) + .max() + .unwrap_or(0), + ), consumer_groups: topic_snap .consumer_groups .into_iter() diff --git a/core/partitions/src/iggy_index_reader.rs b/core/partitions/src/iggy_index_reader.rs index a105a6982..cd618c19f 100644 --- a/core/partitions/src/iggy_index_reader.rs +++ b/core/partitions/src/iggy_index_reader.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use crate::iggy_index::{IGGY_INDEX_SIZE, IggyIndex}; +use crate::iggy_index::{IGGY_INDEX_SIZE, IggyIndex, IggyIndexCache}; use bytes::Buf; use compio::fs::{File, OpenOptions}; use compio::io::AsyncReadAtExt; @@ -114,4 +114,37 @@ impl IggyIndexReader { .await?, )) } + + /// Load every whole entry into an [`IggyIndexCache`] for offset / timestamp + /// lower-bound lookups. Reads the whole file in one pass (index files are + /// tiny: one sparse entry per flushed chunk). A trailing partial entry + /// (torn write) is ignored (see [`Self::entry_count`]). + /// + /// # Errors + /// + /// Returns an error if the file metadata or bytes cannot be read. + pub async fn load_all(&self) -> Result<IggyIndexCache, IggyError> { + let count = usize::try_from(self.entry_count().await?) + .map_err(|_| IggyError::CannotReadFileMetadata)?; + if count == 0 { + return Ok(IggyIndexCache::empty()); + } + + // `with_capacity` (len 0): `read_exact_at` fills the spare capacity in + // place and advances the length (see `read_entry_at`). + let buffer = Vec::with_capacity(count * IGGY_INDEX_SIZE); + let (result, buffer): (std::io::Result<()>, Vec<u8>) = + self.file.read_exact_at(buffer, 0).await.into(); + result.map_err(|_| IggyError::CannotReadFile)?; + + let mut cache = IggyIndexCache::with_capacity(count); + let mut view = buffer.as_slice(); + for _ in 0..count { + let offset = view.get_u64_le(); + let timestamp = view.get_u64_le(); + let position = view.get_u64_le(); + cache.insert(offset, timestamp, position); + } + Ok(cache) + } } diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 8d2c3a4bf..d2256cb70 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -57,7 +57,8 @@ use server_common::{ MESSAGE_ALIGN, Message, SegmentStorage, iobuf::{Frozen, Owned}, send_messages2::{ - convert_request_message, decode_prepare_slice, stamp_prepare_for_persistence, + ChecksumMode, convert_request_message, decode_prepare_slice, decode_prepare_slice_trusted, + stamp_prepare_for_persistence, verify_received_send_messages, }, sharding::IggyNamespace, }; @@ -723,8 +724,9 @@ where /// can run the disk read + offset persist off the partition borrow. The /// in-memory journal tier is read here directly (mem reads never yield); /// the disk tier is captured as owned descriptors in [`DiskReadPlan`]. + #[allow(clippy::too_many_lines)] pub(crate) fn build_poll_plan( - &self, + &mut self, consumer: PollingConsumer, args: &PollingArgs, ) -> PollPlan { @@ -815,13 +817,22 @@ where } let (start_segment, start_position) = self.disk_poll_start(&query); + // Cap resident sealed read handles: touch this poll's start segment so + // the LRU keeps the hot set and drops the least-recently-used fd + index. + let start_offset = self.log.segments()[start_segment].start_offset; + self.log.touch_sealed_read_state(start_offset); // Snapshot only the segments the disk walk visits (`start_segment..`), - // so `start_position` applies to the first snapshotted segment. + // so `start_position` applies to the first snapshotted segment. A sealed + // segment carries its shared read-state handle (fd + sparse index) so + // the off-borrow read reuses (or fills) it; the active segment opens + // fresh and resolves from its resident index. let segments = self.log.segments()[start_segment..] .iter() - .map(|segment| DiskSegment { + .zip(self.log.sealed_read_state()[start_segment..].iter()) + .map(|(segment, read_state)| DiskSegment { start_offset: segment.start_offset, persisted: segment.size.as_bytes_u64(), + read_state: segment.sealed.then(|| Rc::clone(read_state)), }) .collect(); let disk = DiskReadPlan { @@ -1119,7 +1130,13 @@ where ); let message = if message.header().operation == Operation::SendMessages { - match convert_request_message(namespace, message) { + // Skip the batch-checksum pass: on the partition ingest path + // nothing reads it before `stamp_prepare_for_persistence` + // recomputes it over the stamped header. An already-canonical + // batch (native v2, or the plane's pre-encrypt convert output) + // returns early above, so Skip only affects the legacy + // transcode, whose output goes straight to project/stamp. + match convert_request_message(namespace, message, ChecksumMode::Skip) { Ok(message) => message, Err(error) => { emit_partition_diag( @@ -1502,6 +1519,32 @@ where ); } } + // First blob-integrity check on the replicated path. The consensus + // layer never validates the body (PrepareHeader integrity fields are + // inert zeros) and the batch checksum is recomputed locally at stamp, + // so a follower must verify each message's stamp-invariant per-message + // checksum before journaling transit bytes. Follower-only: the primary + // (and single-node self-replicate) produced these bytes and already + // checked the client batch at ingest, so they must not pay this pass. + // Fail closed on mismatch - drop without journaling, forwarding, or + // acking; the primary retransmits on prepare-timeout. + if is_backup + && header.operation == Operation::SendMessages + && let Err(error) = verify_received_send_messages(message.as_slice()) + { + emit_partition_diag( + tracing::Level::WARN, + &PartitionDiagEvent::new( + self.diag_ctx(), + "rejecting replicated send_messages: per-message checksum mismatch", + ) + .with_operation(header.operation) + .with_op(header.op) + .with_error(error.to_string()), + ); + return; + } + // Durability-before-ack: clone for chain-replicate, forward only // AFTER apply_replicated_operation persists. Forward-first would // give downstream an op whose WAL entry we never wrote, that violates @@ -1911,11 +1954,15 @@ where } continue; } - // A resident committed SendMessages entry decoded once at append - // (the offset index) with its checksum stamped over these exact - // bytes, so it must decode again here. Guard the invariant for a - // future disk read-back path that could make decode fallible. - let Ok(batch) = decode_prepare_slice(entry.as_slice()) else { + // Resident committed SendMessages entry: this node stamped it + // in `append_messages` (recomputing the batch checksum over these + // exact bytes), so a validating re-decode would only re-hash ~1 + // MiB to confirm our own write. Trust the structural decode; the + // batch-checksum recompute belongs at network ingress (repair + // validation + the follower receive gate), not on locally-stamped + // bytes. Guard the invariant for a future disk read-back path that + // could make decode fallible. + let Ok(batch) = decode_prepare_slice_trusted(entry.as_slice()) else { tracing::error!( target: "iggy.partitions.diag", namespace_raw = self.namespace().inner(), @@ -2315,8 +2362,11 @@ where let Some(entry) = self.log.journal().inner.entry(prepare_header).await else { return Err(IggyError::InvalidCommand); }; - let batch = - decode_prepare_slice(entry.as_slice()).map_err(|_| IggyError::InvalidCommand)?; + // Trusted (no batch-hash): the entry was read back from this replica's + // own journal, where it was stamped/validated at append; only header + // stats are needed, so re-hashing the ~1 MiB blob is redundant. + let batch = decode_prepare_slice_trusted(entry.as_slice()) + .map_err(|_| IggyError::InvalidCommand)?; let message_count = batch.message_count(); if message_count == 0 { return Ok(None); @@ -2794,6 +2844,10 @@ where self.log.indexes_mut().remove(0); self.log.messages_writers_mut().remove(0); self.log.index_writers_mut().remove(0); + // Drop the pump's read-state handle (fd + sparse index); an in-flight + // poll holding a clone keeps it alive until it finishes (a cached fd + // reads the unlinked inode). + self.log.sealed_read_state_mut().remove(0); let (messages_path, index_path) = storage.segment_and_index_paths(); let _ = storage.shutdown(); @@ -2952,6 +3006,8 @@ where self.log.indexes_mut().remove(0); self.log.messages_writers_mut().remove(0); self.log.index_writers_mut().remove(0); + // Drop the pump's read-state handle in lockstep (see cleanup path). + self.log.sealed_read_state_mut().remove(0); let (messages_path, index_path) = storage.segment_and_index_paths(); let _ = storage.shutdown(); @@ -3433,7 +3489,7 @@ fn nth_oldest_sealed_end(segments: &[Segment], count: u32) -> Option<u64> { #[cfg(test)] mod tests { use super::*; - use crate::poll_plan::DiskReadOutcome; + use crate::poll_plan::{DiskReadOutcome, SealedSegmentHandle}; use bytes::Bytes; use compio::io::AsyncWriteAtExt; use consensus::LocalPipeline; @@ -3889,10 +3945,12 @@ mod tests { DiskSegment { start_offset: 0, persisted: 512, + read_state: None, }, DiskSegment { start_offset: 5, persisted: later_len, + read_state: None, }, ], start_position: 0, @@ -3972,10 +4030,12 @@ mod tests { DiskSegment { start_offset: 0, persisted: corrupt_len, + read_state: None, }, DiskSegment { start_offset: 5, persisted: later_len, + read_state: None, }, ], start_position: 0, @@ -3998,6 +4058,183 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + /// A sealed-segment poll opens the file once and caches the read fd; a later + /// poll of the same segment reuses the cached descriptor. Proven by + /// unlinking the file after the first read: a fresh open-by-path would now + /// fail, so a successful second read can only come from the cached fd (which + /// reads the still-open, unlinked inode). + #[compio::test] + async fn read_disk_caches_and_reuses_sealed_segment_fd() { + let namespace = IggyNamespace::new(1, 1, 0); + + let dir = std::env::temp_dir().join(format!( + "iggy-read-disk-fdcache-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system clock after epoch") + .as_nanos(), + )); + compio::fs::create_dir_all(&dir) + .await + .expect("create temp partition dir"); + let partition_dir = dir.to_string_lossy().into_owned(); + + let record = build_segment_record(namespace, 0); + let record_len = record.len() as u64; + let path = format!("{partition_dir}/{:0>20}.log", 0u64); + { + let mut file = compio::fs::File::create(&path) + .await + .expect("create segment file"); + let (written, _) = file.write_all_at(record, 0).await.into(); + written.expect("write segment record"); + file.sync_all().await.expect("flush segment file"); + } + + let handle = SealedSegmentHandle::default(); + assert!(handle.fd.borrow().is_none(), "fd cache slot starts empty"); + + let plan = DiskReadPlan { + partition_dir: Some(partition_dir.clone()), + segments: vec![DiskSegment { + start_offset: 0, + persisted: record_len, + read_state: Some(Rc::clone(&handle)), + }], + start_position: 0, + namespace_raw: namespace.inner(), + }; + let first = plan + .read_disk(MessageLookup::Offset { + offset: 0, + count: 1, + ceiling: u64::MAX, + }) + .await; + assert!( + matches!(first, DiskReadOutcome::Matched { .. }), + "first sealed poll must match the batch", + ); + assert!( + handle.fd.borrow().is_some(), + "first sealed poll must populate the read-fd cache slot", + ); + + // Unlink the file: a fresh open-by-path would fail now, so the second + // read succeeding proves the cached fd was reused. + std::fs::remove_file(&path).expect("unlink segment file"); + + let plan = DiskReadPlan { + partition_dir: Some(partition_dir.clone()), + segments: vec![DiskSegment { + start_offset: 0, + persisted: record_len, + read_state: Some(Rc::clone(&handle)), + }], + start_position: 0, + namespace_raw: namespace.inner(), + }; + let second = plan + .read_disk(MessageLookup::Offset { + offset: 0, + count: 1, + ceiling: u64::MAX, + }) + .await; + assert!( + matches!(second, DiskReadOutcome::Matched { .. }), + "cached fd must serve the read after the segment path is unlinked", + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + /// A sealed-segment poll reloads the dropped sparse index from the `.index` + /// file and resolves the start byte from it, skipping the full-segment scan. + /// Proven by prefixing the `.log` with bytes a scan from position 0 would + /// fault on: only an index that jumps straight to the batch reads it. + #[compio::test] + async fn read_disk_reloads_sealed_index_to_skip_scan() { + let namespace = IggyNamespace::new(1, 1, 0); + + let dir = std::env::temp_dir().join(format!( + "iggy-read-disk-idxreload-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system clock after epoch") + .as_nanos(), + )); + compio::fs::create_dir_all(&dir) + .await + .expect("create temp partition dir"); + let partition_dir = dir.to_string_lossy().into_owned(); + + // `.log`: an undecodable prefix (a scan from byte 0 faults on it) then a + // valid batch at offset 5. `.index`: one sparse entry mapping offset 5 + // to the batch's byte position, so the poll jumps past the prefix. + let prefix = vec![0xABu8; 512]; + let prefix_len = prefix.len() as u64; + let batch = build_segment_record(namespace, 5); + let mut log_bytes = prefix; + log_bytes.extend_from_slice(&batch); + let log_len = log_bytes.len() as u64; + let log_path = format!("{partition_dir}/{:0>20}.log", 0u64); + { + let mut file = compio::fs::File::create(&log_path) + .await + .expect("create segment log"); + let (written, _) = file.write_all_at(log_bytes, 0).await.into(); + written.expect("write segment log"); + file.sync_all().await.expect("flush segment log"); + } + + let index_bytes = crate::iggy_index::IggyIndexCache::serialize( + &crate::iggy_index::IggyIndex::new(5, 0, prefix_len), + ); + let index_path = format!("{partition_dir}/{:0>20}.index", 0u64); + { + let mut file = compio::fs::File::create(&index_path) + .await + .expect("create segment index"); + let (written, _) = file.write_all_at(index_bytes, 0).await.into(); + written.expect("write segment index"); + file.sync_all().await.expect("flush segment index"); + } + + let handle = SealedSegmentHandle::default(); + let plan = DiskReadPlan { + partition_dir: Some(partition_dir.clone()), + segments: vec![DiskSegment { + start_offset: 0, + persisted: log_len, + read_state: Some(Rc::clone(&handle)), + }], + // Byte 0, exactly what disk_poll_start returns for a sealed segment + // whose resident index was dropped. + start_position: 0, + namespace_raw: namespace.inner(), + }; + let outcome = plan + .read_disk(MessageLookup::Offset { + offset: 5, + count: 1, + ceiling: u64::MAX, + }) + .await; + assert!( + matches!(outcome, DiskReadOutcome::Matched { .. }), + "the reloaded sparse index must skip the prefix; a scan from byte 0 would fault", + ); + assert!( + handle.index.borrow().is_some(), + "the sealed poll must cache the reloaded sparse index", + ); + + let _ = std::fs::remove_dir_all(&dir); + } + fn repair_config() -> PartitionsConfig { PartitionsConfig { messages_required_to_save: 1, diff --git a/core/partitions/src/iggy_partitions.rs b/core/partitions/src/iggy_partitions.rs index 14e7f0fe7..b15f30407 100644 --- a/core/partitions/src/iggy_partitions.rs +++ b/core/partitions/src/iggy_partitions.rs @@ -26,7 +26,7 @@ use iggy_binary_protocol::{ Command2, ConsensusHeader, Operation, PrepareHeader, PrepareOkHeader, RequestHeader, }; use message_bus::MessageBus; -use server_common::send_messages2::{convert_request_message, encrypt_batch_request}; +use server_common::send_messages2::{ChecksumMode, convert_request_message, encrypt_batch_request}; use server_common::sharding::{IggyNamespace, LocalIdx, ShardId}; #[cfg(debug_assertions)] use std::cell::Cell; @@ -391,9 +391,10 @@ where } /// Build an owned [`PollPlan`] for a partition poll synchronously, under a - /// single [`Self::with_partition`] borrow (the in-memory journal tier + the - /// resident-tail straddle snapshot are read here; mem reads never yield). - /// Returns `None` for a missing or tombstoned namespace. + /// single pump-only `&mut` borrow (the in-memory journal tier + the + /// resident-tail straddle snapshot are read here, and the sealed-read-handle + /// LRU is touched; mem reads never yield). Returns `None` for a missing or + /// tombstoned namespace. /// /// Pairs with [`PollPlan::execute`], which runs the disk read + /// offset persist/apply off the borrow on the owned plan. Splitting the @@ -406,9 +407,11 @@ where consumer: PollingConsumer, args: &PollingArgs, ) -> Option<PollPlan> { - self.with_partition(namespace, |partition| { - partition.build_poll_plan(consumer, args) - }) + // `build_poll_plan` touches the partition's sealed-read-handle LRU, so it + // needs `&mut`. Sound on the pump: it is fully synchronous (no `.await` + // inside), so no sibling task can realloc the partitions vec under it. + let partition = self.get_mut_by_ns(namespace)?; + Some(partition.build_poll_plan(consumer, args)) } /// Read a consumer's stored offset + the partition commit offset. Fully @@ -509,7 +512,11 @@ where let message = if message.header().operation == Operation::SendMessages && let Some(encryptor) = &self.config().encryptor { - let canonical = convert_request_message(namespace, message) + // Compute the batch checksum: this canonical output is validated by + // `encrypt_batch_request`'s decode before re-encryption, and the + // re-encrypted batch (checksum kept by `encrypt_batch_request`) then + // re-enters `convert` as the canonical-vs-legacy discriminator. + let canonical = convert_request_message(namespace, message, ChecksumMode::Compute) .and_then(|message| encrypt_batch_request(message, encryptor)); match canonical { Ok(message) => message, diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs index feb78c24b..e4b05d065 100644 --- a/core/partitions/src/journal.rs +++ b/core/partitions/src/journal.rs @@ -19,7 +19,9 @@ use iggy_binary_protocol::{Operation, PrepareHeader}; use journal::{Journal, Storage}; use server_common::{ iobuf::{Frozen, Owned}, - send_messages2::{COMMAND_HEADER_SIZE, SendMessages2Ref, decode_prepare_slice}, + send_messages2::{ + COMMAND_HEADER_SIZE, SendMessages2Ref, decode_prepare_slice, decode_prepare_slice_trusted, + }, }; use std::io; use std::{ @@ -519,8 +521,12 @@ impl PartitionJournal<PartitionJournalMemStorage> { // One decode feeds both the offset/timestamp index (keyed on // `origin_timestamp`) and the surfaced accounting meta (`base_timestamp`, // size, count); the two timestamps are distinct fields, do not conflate. + // Trusted (no batch-hash): every entry reaching append was just stamped + // by `stamp_prepare_for_persistence` (its checksum recomputed over this + // exact blob) or re-appended from an already-validated resident entry, + // so re-hashing the ~1 MiB blob here only to read the header is waste. let (index_offset_timestamp, meta) = if header.operation == Operation::SendMessages { - match decode_prepare_slice(entry.as_slice()) { + match decode_prepare_slice_trusted(entry.as_slice()) { Ok(batch) if batch.message_count() != 0 => { let message_count = batch.message_count(); let meta = RetainedBatchMeta { diff --git a/core/partitions/src/log.rs b/core/partitions/src/log.rs index 4fd5673db..182425f90 100644 --- a/core/partitions/src/log.rs +++ b/core/partitions/src/log.rs @@ -18,11 +18,13 @@ use crate::iggy_index::{IGGY_INDEX_SIZE, IggyIndexCache}; use crate::iggy_index_writer::IggyIndexWriter; use crate::messages_writer::MessagesWriter; +use crate::poll_plan::SealedSegmentHandle; use crate::segment::Segment; use iggy_common::{IggyByteSize, IggyMessagesBatch}; use journal::{Journal, Storage}; use ringbuffer::AllocRingBuffer; use server_common::{IggyMessagesBatchSetInFlight, SegmentStorage}; +use std::collections::VecDeque; use std::fmt::Debug; use std::rc::Rc; @@ -30,6 +32,14 @@ const SEGMENTS_CAPACITY: usize = 1024; const ACCESS_MAP_CAPACITY: usize = 8; const SIZE_16MB: usize = 16 * 1024 * 1024; +/// Max sealed segments per partition that keep a resident read handle (fd + +/// sparse index). Without a cap every sealed segment a reader ever touched pins +/// one fd for the partition's lifetime; the server-wide budget is this cap times +/// the partition count, so keep it small. 12 covers a lagging consumer's working +/// set (the recent sealed segments it re-reads) with room for a few concurrent +/// readers before an LRU eviction forces a re-open. +const SEALED_READ_STATE_CAP: usize = 12; + /// Tracking metadata for the journal's current state. /// /// Replaces the server journal's `Inner` struct — lives in the `SegmentedLog` @@ -112,6 +122,16 @@ where storage: Vec<SegmentStorage>, messages_writers: Vec<Option<Rc<MessagesWriter>>>, index_writers: Vec<Option<Rc<IggyIndexWriter>>>, + // Parallel to `segments`: a shared read-state handle (fd + sparse index) + // per segment, filled lazily on the first sealed-segment poll and cloned + // into the off-borrow poll plan. Maintained in lockstep with `segments` + // (push/remove together). + sealed_read_state: Vec<SealedSegmentHandle>, + // LRU of sealed-segment `start_offset`s (most-recently-used at the front) + // bounding how many `sealed_read_state` handles stay resident, capped at + // `SEALED_READ_STATE_CAP`. Keyed by offset (stable), not slot index (which + // shifts on retire). See `touch_sealed_read_state`. + sealed_lru: VecDeque<u64>, in_flight: IggyMessagesBatchSetInFlight, } @@ -131,6 +151,8 @@ where indexes: Vec::with_capacity(SEGMENTS_CAPACITY), messages_writers: Vec::with_capacity(SEGMENTS_CAPACITY), index_writers: Vec::with_capacity(SEGMENTS_CAPACITY), + sealed_read_state: Vec::with_capacity(SEGMENTS_CAPACITY), + sealed_lru: VecDeque::with_capacity(SEALED_READ_STATE_CAP + 1), in_flight: IggyMessagesBatchSetInFlight::default(), } } @@ -153,6 +175,49 @@ where &mut self.segments } + /// Shared read-state handles, parallel to [`Self::segments`]. Cloned into + /// the poll plan for sealed segments (see [`SealedSegmentHandle`]). + pub const fn sealed_read_state(&self) -> &Vec<SealedSegmentHandle> { + &self.sealed_read_state + } + + /// Mutable read-state handles. `remove(0)` in lockstep with `segments` when + /// a segment is retired so the pump drops its handle (freeing the fd + + /// index once any in-flight poll holding a clone completes). + pub const fn sealed_read_state_mut(&mut self) -> &mut Vec<SealedSegmentHandle> { + &mut self.sealed_read_state + } + + /// Record a sealed-segment access and enforce [`SEALED_READ_STATE_CAP`] + /// (LRU). `start_offset` keys the segment - stable across retire, unlike the + /// slot index. It moves to the most-recently-used front; once more than the + /// cap distinct sealed segments are tracked, the least-recently-used one's + /// handle is dropped (replaced with a fresh empty handle) so its fd + sparse + /// index free. An in-flight poll holding a clone of the dropped handle keeps + /// it alive until it finishes (see [`SealedSegmentHandle`]). + pub fn touch_sealed_read_state(&mut self, start_offset: u64) { + if let Some(pos) = self + .sealed_lru + .iter() + .position(|&offset| offset == start_offset) + { + self.sealed_lru.remove(pos); + } + self.sealed_lru.push_front(start_offset); + if self.sealed_lru.len() > SEALED_READ_STATE_CAP { + let Some(evicted) = self.sealed_lru.pop_back() else { + return; + }; + if let Some(slot) = self + .segments + .iter() + .position(|segment| segment.start_offset == evicted) + { + self.sealed_read_state[slot] = SealedSegmentHandle::default(); + } + } + } + pub const fn storages_mut(&mut self) -> &mut Vec<SegmentStorage> { &mut self.storage } @@ -259,6 +324,7 @@ where self.indexes.push(None); self.messages_writers.push(messages_writer); self.index_writers.push(index_writer); + self.sealed_read_state.push(SealedSegmentHandle::default()); } pub fn set_segment_indexes(&mut self, segment_index: usize, indexes: IggyIndexCache) { @@ -297,3 +363,114 @@ where &self.journal } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::journal::{PartitionJournal, PartitionJournalMemStorage}; + + type TestLog = + SegmentedLog<PartitionJournal<PartitionJournalMemStorage>, PartitionJournalMemStorage>; + + /// Push a sealed segment with a resident (index-filled) read handle and + /// return a clone of that handle, standing in for an in-flight poll's clone. + fn push_resident_sealed(log: &mut TestLog, start_offset: u64) -> SealedSegmentHandle { + log.segments.push(Segment { + start_offset, + sealed: true, + ..Segment::default() + }); + log.sealed_read_state.push(SealedSegmentHandle::default()); + let slot = log.sealed_read_state.len() - 1; + *log.sealed_read_state[slot].index.borrow_mut() = Some(IggyIndexCache::with_capacity(1)); + Rc::clone(&log.sealed_read_state[slot]) + } + + #[test] + fn touch_sealed_read_state_evicts_least_recently_used_past_cap() { + let mut log = TestLog::default(); + let handles: Vec<_> = (0..=SEALED_READ_STATE_CAP as u64) + .map(|offset| push_resident_sealed(&mut log, offset)) + .collect(); + // Ascending touch order: offset 0 is the least-recently used. + for offset in 0..=SEALED_READ_STATE_CAP as u64 { + log.touch_sealed_read_state(offset); + } + + // Slot 0 dropped: the pump handle was replaced with a fresh empty one. + assert!( + !Rc::ptr_eq(&handles[0], &log.sealed_read_state()[0]), + "least-recently-used handle must be dropped past the cap", + ); + assert!( + log.sealed_read_state()[0].index.borrow().is_none(), + "the dropped slot resets to an empty handle", + ); + // In-flight safety: the dropped handle's clone stays alive and still + // sees its cached index, so a poll holding it finishes without a UAF. + assert_eq!( + Rc::strong_count(&handles[0]), + 1, + "the dropped handle survives for an in-flight poll's clone", + ); + assert!( + handles[0].index.borrow().is_some(), + "the in-flight clone keeps reading the cached index", + ); + // Every more-recently-used slot is retained (same Rc). + for (handle, resident) in handles.iter().zip(log.sealed_read_state().iter()).skip(1) { + assert!( + Rc::ptr_eq(handle, resident), + "recently-used handles stay resident", + ); + } + } + + #[test] + fn touch_sealed_read_state_reorders_eviction_on_reaccess() { + let mut log = TestLog::default(); + let mut handles: Vec<_> = (0..SEALED_READ_STATE_CAP as u64) + .map(|offset| push_resident_sealed(&mut log, offset)) + .collect(); + for offset in 0..SEALED_READ_STATE_CAP as u64 { + log.touch_sealed_read_state(offset); + } + // Re-access offset 0 -> now most-recently used, so offset 1 becomes the + // least-recently used and next to be evicted. + log.touch_sealed_read_state(0); + + let new_offset = SEALED_READ_STATE_CAP as u64; + handles.push(push_resident_sealed(&mut log, new_offset)); + log.touch_sealed_read_state(new_offset); + + assert!( + !Rc::ptr_eq(&handles[1], &log.sealed_read_state()[1]), + "the least-recently-used segment is evicted, not the re-accessed one", + ); + assert!( + Rc::ptr_eq(&handles[0], &log.sealed_read_state()[0]), + "the re-accessed segment stays resident", + ); + } + + #[test] + fn evicted_sealed_slot_is_empty_and_refillable() { + let mut log = TestLog::default(); + for offset in 0..=SEALED_READ_STATE_CAP as u64 { + push_resident_sealed(&mut log, offset); + } + for offset in 0..=SEALED_READ_STATE_CAP as u64 { + log.touch_sealed_read_state(offset); + } + + // The evicted slot holds a fresh empty handle, so the next poll re-opens + // instead of reusing a stale descriptor. + let evicted = &log.sealed_read_state()[0]; + assert!(evicted.fd.borrow().is_none()); + assert!(evicted.index.borrow().is_none()); + + // Re-filling it (what the next sealed poll does) works. + *log.sealed_read_state()[0].index.borrow_mut() = Some(IggyIndexCache::with_capacity(1)); + assert!(log.sealed_read_state()[0].index.borrow().is_some()); + } +} diff --git a/core/partitions/src/poll_plan.rs b/core/partitions/src/poll_plan.rs index 4792760d3..a35a1a8e4 100644 --- a/core/partitions/src/poll_plan.rs +++ b/core/partitions/src/poll_plan.rs @@ -24,11 +24,15 @@ //! synchronously under the borrow into the owned types here, drops the borrow, //! then [`PollPlan::execute`] runs the disk read + the in-memory auto-commit //! apply on owned data alone: consumer offsets are already `Arc`, the journal -//! tail is a point-in-time `Frozen` snapshot, and segment files are re-opened -//! by path. No value in this module holds a partition reference, so executing a -//! plan is sound on a detached task concurrently with the pump's own writes. +//! tail is a point-in-time `Frozen` snapshot, and each sealed segment carries a +//! shared [`SealedSegmentReadState`] handle (a plain `Rc`, not a partition +//! reference) whose read fd + sparse index the read reuses or fills on a miss. +//! No value in this module holds a partition reference, so executing a plan is +//! sound on a detached task concurrently with the pump's own writes. use crate::PollFragments; +use crate::iggy_index::IggyIndexCache; +use crate::iggy_index_reader::IggyIndexReader; use crate::journal::{MessageLookup, push_selected_batch_fragments, select_batch_slice}; use compio::io::AsyncReadAtExt; use iggy_common::{ @@ -36,14 +40,39 @@ use iggy_common::{ }; use server_common::iobuf::{Frozen, Owned}; use server_common::send_messages2::{COMMAND_HEADER_SIZE, decode_batch_slice}; +use std::cell::RefCell; use std::hash::Hash; +use std::rc::Rc; use std::sync::Arc; use std::sync::atomic::Ordering; use tracing::warn; -/// Owned, borrow-free inputs for the disk tier of a poll (see module docs). -/// Segment files are re-opened by path because sealed segments drop their -/// writer at rotation. +/// Per-sealed-segment read state, shared as a cheap `Rc` handle between the +/// owning partition and the off-borrow [`DiskReadPlan`] (a plain `Rc`, never a +/// partition reference, so the read runs off the pump). Both slots fill lazily +/// on the first sealed poll and are reused after. The pump drops its handle +/// when the segment is retired, so the state frees once any in-flight poll +/// holding a clone finishes (a cached fd meanwhile reads the unlinked inode, +/// which is fine). The active segment is never cached. +#[derive(Debug, Default)] +pub struct SealedSegmentReadState { + /// Read-only descriptor; compio `File` clones share the kernel fd, so a hit + /// avoids the per-poll `openat` (an `io_uring` op prone to io-wq punts) and + /// preserves kernel readahead. `None` until the first sealed poll opens it. + pub(crate) fd: RefCell<Option<compio::fs::File>>, + /// Sparse offset/timestamp index reloaded from the `.index` file the + /// segment dropped at rotation, so a poll resolves the start byte in + /// O(log n) instead of scanning the whole segment from byte 0 (the stall). + /// `None` until the first sealed poll loads it. + pub(crate) index: RefCell<Option<IggyIndexCache>>, +} + +pub type SealedSegmentHandle = Rc<SealedSegmentReadState>; + +/// Owned, borrow-free inputs for the disk tier of a poll (see module docs). A +/// sealed segment reuses its cached [`SealedSegmentReadState`] (read fd + sparse +/// index); the active segment (and any cache miss) opens by path and resolves +/// from its resident index, because sealed segments drop both at rotation. pub struct DiskReadPlan { pub(crate) partition_dir: Option<String>, /// Segments to walk, snapshotted from the poll's starting segment onward @@ -57,6 +86,10 @@ pub struct DiskReadPlan { pub struct DiskSegment { pub(crate) start_offset: u64, pub(crate) persisted: u64, + /// Shared read state, cloned from the owning partition at plan time for a + /// SEALED segment; `None` for the active segment, which always opens fresh + /// and resolves from its resident index. See [`SealedSegmentReadState`]. + pub(crate) read_state: Option<SealedSegmentHandle>, } /// Owned auto-commit input, applied off the partition borrow after a poll (see @@ -406,7 +439,21 @@ impl DiskReadPlan { // `start_position` applies to the first snapshotted segment; each later // segment is walked from byte 0 (reset at the end of every iteration). - let mut position = self.start_position; + // + // A sealed first segment dropped its resident index at rotation, so + // `disk_poll_start` fell back to byte 0. Reload the sparse index (once, + // then cached) and resolve the start byte so the walk skips straight to + // the target instead of scanning the whole segment - the poll stall. A + // miss or load failure keeps `start_position` (the pre-existing + // full-scan fallback). The active segment carries no read state, so its + // resident-index-resolved `start_position` is left untouched. + let mut position = match self.segments.first() { + Some(first) => self + .resolve_sealed_start(first, query, partition_dir) + .await + .unwrap_or(self.start_position), + None => self.start_position, + }; let mut fragments = PollFragments::new(); let mut last_matching_offset = None; let mut matched: u32 = 0; @@ -427,7 +474,7 @@ impl DiskReadPlan { continue; } let path = format!("{partition_dir}/{:0>20}.log", segment.start_offset); - let Some(file) = self.open_segment_with_retry(&path).await else { + let Some(file) = self.resolve_segment_file(segment, &path).await else { // Open exhausted retries: the segment may hold present-but- // unreadable data. Stop here rather than walking past it. faulted = true; @@ -486,6 +533,94 @@ impl DiskReadPlan { } } + /// Resolve the read-only descriptor for `segment`'s file. A sealed segment + /// clones its cached fd on a hit (sharing the kernel fd, no syscall) and, on + /// a miss, opens by path and stores the fd back so later polls skip the + /// `openat`. The active segment (no cache slot) always opens fresh. Returns + /// `None` only when the open exhausts its retries (the caller fails closed). + async fn resolve_segment_file( + &self, + segment: &DiskSegment, + path: &str, + ) -> Option<compio::fs::File> { + let Some(handle) = &segment.read_state else { + return self.open_segment_with_retry(path).await; + }; + // Borrow only to clone the `Option<File>` out, never across the await. + if let Some(cached) = handle.fd.borrow().clone() { + return Some(cached); + } + let file = self.open_segment_with_retry(path).await?; + // Benign race: a concurrent poll of the same segment may have filled the + // slot while this open was in flight; overwriting with an equivalent fd + // (same inode) is harmless. + *handle.fd.borrow_mut() = Some(file.clone()); + Some(file) + } + + /// Resolve the start byte for the poll's target segment from its sparse + /// index, loading the `.index` file on the first sealed poll and caching it + /// on the shared handle. Returns `None` (keep the byte-0 fallback) for the + /// active segment (no handle), a below-range query, or a load failure. + async fn resolve_sealed_start( + &self, + segment: &DiskSegment, + query: MessageLookup, + partition_dir: &str, + ) -> Option<u64> { + // TODO: a per-consumer cursor hint (the previous sealed poll's resolved + // position) could seed this so a sequentially advancing consumer skips + // the sparse-index lookup on repeated polls of the same segment. + let handle = segment.read_state.as_ref()?; + // Cache hit: resolve under a short borrow, never across the await. + let cached = handle + .index + .borrow() + .as_ref() + .map(|index| resolve_index_position(index, query)); + if let Some(resolved) = cached { + return resolved; + } + let path = format!("{partition_dir}/{:0>20}.index", segment.start_offset); + let index = self.load_sealed_index(&path).await?; + let resolved = resolve_index_position(&index, query); + *handle.index.borrow_mut() = Some(index); + resolved + } + + /// Load a sealed segment's sparse index from its `.index` file. `None` on a + /// missing/unreadable file so the caller falls back to a byte-0 scan (the + /// pre-existing behavior); the load is retried on the next poll. + async fn load_sealed_index(&self, path: &str) -> Option<IggyIndexCache> { + match IggyIndexReader::new(path).await { + Ok(reader) => match reader.load_all().await { + Ok(index) => Some(index), + Err(error) => { + warn!( + target: "iggy.partitions.diag", + plane = "partitions", + namespace_raw = self.namespace_raw, + path, + %error, + "disk poll: failed to read sparse index; scanning from segment start" + ); + None + } + }, + Err(error) => { + warn!( + target: "iggy.partitions.diag", + plane = "partitions", + namespace_raw = self.namespace_raw, + path, + %error, + "disk poll: failed to open sparse index; scanning from segment start" + ); + None + } + } + } + /// Open a segment file for a disk poll, retrying transient IO failures (fd /// pressure under heavy parallel load) so one failed syscall does not /// silently collapse the poll into an empty result. @@ -544,6 +679,18 @@ impl DiskReadPlan { } } +/// Byte position of the sparse-index entry at or below the query's offset / +/// timestamp, or `None` when the query is below the first indexed entry (the +/// caller then scans from the segment start). Mirrors `disk_poll_start`'s +/// resident-index resolution for the sealed, off-pump path. +fn resolve_index_position(index: &IggyIndexCache, query: MessageLookup) -> Option<u64> { + match query { + MessageLookup::Offset { offset, .. } => index.offset_lower_bound(offset), + MessageLookup::Timestamp { timestamp, .. } => index.timestamp_lower_bound(timestamp), + } + .map(|entry| entry.position) +} + impl AutoCommitCtx { /// The offset key (kind + numeric id) this auto-commit targets, for the /// replicated `StoreConsumerOffset2` op the serving shard submits. diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml index 7a185a80d..3bae43290 100644 --- a/core/server-ng/config.toml +++ b/core/server-ng/config.toml @@ -668,8 +668,7 @@ ports = { tcp = 8091, quic = 8081, http = 3001, websocket = 8093, tcp_replica = # - numa settings: # + "numa:auto": Use all available numa node, cores # + "numa:nodes=0,1;cores=4;no_ht=true": Use NUMA node 0 and 1, each nodes use 4 cores, and no hyperthreads -# TODO(hubcio): revert to "numa:auto" once multi-shard server-ng is stable. -cpu_allocation = 1 +cpu_allocation = "numa:auto" # Whether shard threads are pinned to dedicated CPU cores (default: true). # Pinned cores are drawn from the process's allowed CPU set (affinity/cpuset diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs index b7d578a3f..eef75a87a 100644 --- a/core/server-ng/src/bootstrap.rs +++ b/core/server-ng/src/bootstrap.rs @@ -481,6 +481,7 @@ struct TcpTopology { replica_listen_addr: Option<SocketAddr>, ws_listen_addr: Option<SocketAddr>, quic_listen_addr: Option<SocketAddr>, + http_listen_addr: Option<SocketAddr>, tcp_tls_listen_addr: Option<SocketAddr>, peers: Vec<(u8, SocketAddr)>, } @@ -1473,19 +1474,13 @@ fn spawn_shutdown_watchdog( /// Copy the configured cluster roster plus this node's own client ports into /// the shared [`ClusterRoster`] so the binary `GetClusterMetadata` read serves /// the real topology. `self_*` back only the cluster-disabled self-synthesis; -/// the HTTP port is read from config since HTTP binds outside this topology. +/// every self port comes from the resolved topology so the reported HTTP port +/// matches the address this node actually binds (the roster port in a cluster). fn build_cluster_roster( config: &ServerNgConfig, topology: &TcpTopology, metadata_view: Arc<AtomicU64>, ) -> ClusterRoster { - let http_port = if config.http.enabled { - parse_socket_addr("http.address", &config.http.address) - .ok() - .map(|addr| addr.port()) - } else { - None - }; ClusterRoster { enabled: config.cluster.enabled, name: config.cluster.name.clone(), @@ -1494,7 +1489,7 @@ fn build_cluster_roster( self_ports: configs::ng_cluster::TransportPorts { tcp: Some(topology.client_listen_addr.port()), quic: topology.quic_listen_addr.map(|addr| addr.port()), - http: http_port, + http: topology.http_listen_addr.map(|addr| addr.port()), websocket: topology.ws_listen_addr.map(|addr| addr.port()), tcp_replica: None, }, @@ -2026,6 +2021,8 @@ fn resolve_tcp_topology( )?; let default_quic_addr = resolve_optional_listener_addr(config.quic.enabled, "quic.address", &config.quic.address)?; + let default_http_addr = + resolve_optional_listener_addr(config.http.enabled, "http.address", &config.http.address)?; if !config.cluster.enabled { if let Some(replica_id) = current_replica_id && replica_id != SHARD_REPLICA_ID @@ -2048,6 +2045,7 @@ fn resolve_tcp_topology( replica_listen_addr: Some(SocketAddr::new(default_client_addr.ip(), 0)), ws_listen_addr: default_ws_addr, quic_listen_addr: default_quic_addr, + http_listen_addr: default_http_addr, tcp_tls_listen_addr: config.tcp.tls.enabled.then_some(default_client_addr), peers: Vec::new(), }); @@ -2068,11 +2066,17 @@ fn resolve_tcp_topology( count: config.cluster.nodes.len(), } })?; - let (client_listen_addr, ws_listen_addr, quic_listen_addr) = resolve_cluster_client_addrs( + let ClusterClientAddrs { + client: client_listen_addr, + ws: ws_listen_addr, + quic: quic_listen_addr, + http: http_listen_addr, + } = resolve_cluster_client_addrs( self_node, default_client_addr, default_ws_addr, default_quic_addr, + default_http_addr, )?; let replica_port = self_node @@ -2096,6 +2100,7 @@ fn resolve_tcp_topology( replica_listen_addr, ws_listen_addr, quic_listen_addr, + http_listen_addr, tcp_tls_listen_addr: config.tcp.tls.enabled.then_some(client_listen_addr), peers, }) @@ -2112,31 +2117,52 @@ fn resolve_optional_listener_addr( Ok(None) } +/// Client-facing listener addresses resolved for this cluster node. Each port +/// comes from the node's roster entry, falling back to the top-level listener +/// default when the roster leaves it unset. +struct ClusterClientAddrs { + client: SocketAddr, + ws: Option<SocketAddr>, + quic: Option<SocketAddr>, + http: Option<SocketAddr>, +} + fn resolve_cluster_client_addrs( self_node: &configs::ng_cluster::ClusterNodeConfig, default_client_addr: SocketAddr, default_ws_addr: Option<SocketAddr>, default_quic_addr: Option<SocketAddr>, -) -> Result<(SocketAddr, Option<SocketAddr>, Option<SocketAddr>), ServerNgError> { + default_http_addr: Option<SocketAddr>, +) -> Result<ClusterClientAddrs, ServerNgError> { let client_port = self_node .ports .tcp .unwrap_or_else(|| default_client_addr.port()); - let client_listen_addr = - socket_addr_from_parts("cluster.nodes[*].ports.tcp", &self_node.ip, client_port)?; - let ws_listen_addr = resolve_cluster_optional_addr( + let client = socket_addr_from_parts("cluster.nodes[*].ports.tcp", &self_node.ip, client_port)?; + let ws = resolve_cluster_optional_addr( self_node, "cluster.nodes[*].ports.websocket", default_ws_addr, |ports| ports.websocket, )?; - let quic_listen_addr = resolve_cluster_optional_addr( + let quic = resolve_cluster_optional_addr( self_node, "cluster.nodes[*].ports.quic", default_quic_addr, |ports| ports.quic, )?; - Ok((client_listen_addr, ws_listen_addr, quic_listen_addr)) + let http = resolve_cluster_optional_addr( + self_node, + "cluster.nodes[*].ports.http", + default_http_addr, + |ports| ports.http, + )?; + Ok(ClusterClientAddrs { + client, + ws, + quic, + http, + }) } fn resolve_cluster_optional_addr( @@ -2208,8 +2234,7 @@ async fn start_tcp_runtime( // HTTP is served over TCP but sits outside the replica_io / manual client // reactor, so it binds independently. Shard-0 gating comes from the sole // caller of this function. - if config.http.enabled { - let http_addr = parse_socket_addr("http.address", &config.http.address)?; + if let Some(http_addr) = topology.http_listen_addr { let self_ports = configs::ng_cluster::TransportPorts { tcp: config .tcp @@ -3303,4 +3328,63 @@ mod tests { "expected MetadataHandoffAborted, got {err:?}" ); } + + fn cluster_node(ip: &str, http: Option<u16>) -> configs::ng_cluster::ClusterNodeConfig { + configs::ng_cluster::ClusterNodeConfig { + name: "node".to_owned(), + ip: ip.to_owned(), + replica_id: 0, + ports: configs::ng_cluster::TransportPorts { + http, + ..Default::default() + }, + } + } + + fn addr(value: &str) -> SocketAddr { + value.parse().expect("valid socket address literal") + } + + #[test] + fn cluster_http_addr_prefers_roster_port_over_default() { + // A byte-identical top-level [http].address is shared across nodes on + // one host; the per-node roster port must win so each binds a distinct + // HTTP socket instead of colliding on the shared default. + let node = cluster_node("127.0.0.1", Some(18090)); + let addrs = resolve_cluster_client_addrs( + &node, + addr("127.0.0.1:18070"), + None, + None, + Some(addr("127.0.0.1:3000")), + ) + .expect("cluster address resolution must succeed"); + assert_eq!(addrs.http, Some(addr("127.0.0.1:18090"))); + } + + #[test] + fn cluster_http_addr_falls_back_to_default_port_on_self_node_ip() { + // No roster HTTP port: keep the top-level port but still bind the + // node's own roster IP, exactly like the ws/quic fallback. + let node = cluster_node("10.0.0.5", None); + let addrs = resolve_cluster_client_addrs( + &node, + addr("10.0.0.5:18070"), + None, + None, + Some(addr("127.0.0.1:3000")), + ) + .expect("cluster address resolution must succeed"); + assert_eq!(addrs.http, Some(addr("10.0.0.5:3000"))); + } + + #[test] + fn cluster_http_addr_is_none_when_http_disabled() { + // http.enabled = false collapses default_http_addr to None; no roster + // port can revive a listener the operator turned off. + let node = cluster_node("127.0.0.1", Some(18090)); + let addrs = resolve_cluster_client_addrs(&node, addr("127.0.0.1:18070"), None, None, None) + .expect("cluster address resolution must succeed"); + assert_eq!(addrs.http, None); + } } diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs index 89160a63a..160f38036 100644 --- a/core/server-ng/src/dispatch.rs +++ b/core/server-ng/src/dispatch.rs @@ -185,10 +185,10 @@ where { let shard_handle = Rc::clone(shard_handle); // Runs synchronously on the shard pump (see `process_lifecycle` -> - // `on_partition_read`). `build_poll_snapshot` takes the partition borrow via - // `with_partition` (closure-scoped, debug `BorrowGuard`) and returns an owned - // `PollPlan`; only owned data crosses into `spawn_poll_io`. A fully-resident - // poll replies here without spawning. See the `poll_plan` module docs. + // `on_partition_read`). `build_poll_snapshot` takes a pump-only `&mut` + // partition borrow (synchronous, so no sibling task can realloc under it) and + // returns an owned `PollPlan`; only owned data crosses into `spawn_poll_io`. A + // fully-resident poll replies here without spawning. See the `poll_plan` module docs. Rc::new(move |namespace, read, reply| { let Some(shard) = upgrade_shard_handle(&shard_handle) else { return; diff --git a/core/server-ng/src/partition_reconciler.rs b/core/server-ng/src/partition_reconciler.rs index 37a94b32f..71968aa34 100644 --- a/core/server-ng/src/partition_reconciler.rs +++ b/core/server-ng/src/partition_reconciler.rs @@ -1759,15 +1759,15 @@ mod tests { let ns = IggyNamespace::new(0, 0, 0); assert!(shard.plane.partitions().contains(&ns)); - // Two groups: "dead" gets id 1, "live" gets id 2 (per-topic monotonic). + // Two groups: "dead" gets id 0, "live" gets id 1 (per-topic monotonic). let stm = &shard.plane.metadata().mux_stm; seed_create_consumer_group(stm, 3, 0, 0, "dead"); seed_create_consumer_group(stm, 4, 0, 0, "live"); // Offsets are keyed by the monotonic group id (the id the store path is // rewritten to and the read path / live-set resolve), not the name hash. - let dead_key: u32 = 1; - let live_key: u32 = 2; + let dead_key: u32 = 0; + let live_key: u32 = 1; { let partitions = shard.plane.partitions(); let partition = partitions.get_by_ns(&ns).expect("partition materialised"); @@ -1781,8 +1781,8 @@ mod tests { ); } - // Delete the "dead" group (id 1); "live" (id 2) stays. - seed_delete_consumer_group(stm, 5, 0, 0, 1); + // Delete the "dead" group (id 0); "live" (id 1) stays. + seed_delete_consumer_group(stm, 5, 0, 0, 0); reconcile_pass(&ctx).await; let partitions = shard.plane.partitions(); @@ -1815,10 +1815,10 @@ mod tests { vec![assignment(0, 1), assignment(1, 2)], ); seed_create_consumer_group(&mux, 3, 0, 0, "cg"); - // Single member owns every partition (group id 1, the first in topic). - seed_join_consumer_group(&mux, 4, 0, 0, 1, 100); + // Single member owns every partition (group id 0, the first in topic). + seed_join_consumer_group(&mux, 4, 0, 0, 0, 100); - let group = WireIdentifier::numeric(1); + let group = WireIdentifier::numeric(0); let stream = WireIdentifier::numeric(0); let topic = WireIdentifier::numeric(0); let assigned = |mux: &TestMux| -> Vec<u32> { @@ -1888,13 +1888,13 @@ mod tests { "topic-dc", vec![assignment(0, 1), assignment(1, 2)], ); - seed_create_consumer_group(&mux, 3, 0, 0, "cg"); // group id 1 - seed_join_consumer_group(&mux, 4, 0, 0, 1, 100); - seed_join_consumer_group(&mux, 5, 0, 0, 1, 200); + seed_create_consumer_group(&mux, 3, 0, 0, "cg"); // group id 0 + seed_join_consumer_group(&mux, 4, 0, 0, 0, 100); + seed_join_consumer_group(&mux, 5, 0, 0, 0, 200); let stream = WireIdentifier::numeric(0); let topic = WireIdentifier::numeric(0); - let group = WireIdentifier::numeric(1); + let group = WireIdentifier::numeric(0); let assigned = |client: u128| -> Option<Vec<u32>> { mux.streams() .consumer_group_member_assignment(&stream, &topic, &group, client) diff --git a/core/server-ng/src/responses.rs b/core/server-ng/src/responses.rs index 9cf3378ed..4d07cef55 100644 --- a/core/server-ng/src/responses.rs +++ b/core/server-ng/src/responses.rs @@ -1432,7 +1432,7 @@ where /// Size of the in-storage (`IggyMessage2`) per-message header inside a /// `SendMessages2` batch blob: `checksum`(8) + `id`(16) + `offset_delta`(4) /// + `timestamp_delta`(4) + `user_headers_length`(4) + `payload_length`(4) -/// + reserved(8). See `server_common::send_messages2::from_legacy_request`. +/// + reserved(8). See `server_common::send_messages2::SendMessages2Owned::from_messages`. const STORED_MESSAGE_HEADER_SIZE: usize = 48; /// Build the `PolledMessages` reply body from the owning shard's poll diff --git a/core/server_common/src/send_messages2.rs b/core/server_common/src/send_messages2.rs index a64cd0664..7e5b29983 100644 --- a/core/server_common/src/send_messages2.rs +++ b/core/server_common/src/send_messages2.rs @@ -194,78 +194,6 @@ impl SendMessages2Owned { Ok(Self { header, blob }) } - pub fn from_legacy_request(namespace: IggyNamespace, body: &[u8]) -> Result<Self, 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; - let mut cursor = 0usize; - - while cursor < messages.len() && parsed.len() < message_count as usize { - let legacy = LegacyMessageRef::decode(&messages[cursor..])?; - origin_timestamp = origin_timestamp.min(legacy.origin_timestamp); - cursor += legacy.total_size; - parsed.push(legacy); - } - - if parsed.len() != message_count as usize || cursor != messages.len() { - return Err(IggyError::InvalidCommand); - } - - if origin_timestamp == u64::MAX { - origin_timestamp = 0; - } - - let mut blob = BytesMut::with_capacity(messages.len()); - for (index, legacy) in parsed.iter().enumerate() { - let id = if legacy.id == 0 { - random_id::get_uuid() - } else { - legacy.id - }; - let offset_delta = u32::try_from(index).map_err(|_| IggyError::InvalidCommand)?; - let timestamp_delta = legacy - .origin_timestamp - .checked_sub(origin_timestamp) - .ok_or(IggyError::InvalidCommand)?; - if timestamp_delta > MAX_TIMESTAMP_DELTA_MICROS { - return Err(IggyError::InvalidMessageTimestampDelta(timestamp_delta)); - } - let timestamp_delta = - u32::try_from(timestamp_delta).map_err(|_| IggyError::InvalidCommand)?; - let user_headers_length = - u32::try_from(legacy.user_headers.len()).map_err(|_| IggyError::InvalidCommand)?; - let payload_length = - u32::try_from(legacy.payload.len()).map_err(|_| IggyError::InvalidCommand)?; - - let mut header = [0u8; MESSAGE_HEADER_SIZE]; - header[8..24].copy_from_slice(&id.to_le_bytes()); - header[24..28].copy_from_slice(&offset_delta.to_le_bytes()); - header[28..32].copy_from_slice(×tamp_delta.to_le_bytes()); - header[32..36].copy_from_slice(&user_headers_length.to_le_bytes()); - header[36..40].copy_from_slice(&payload_length.to_le_bytes()); - - let checksum = - calculate_checksum_parts(&header[8..], legacy.payload, legacy.user_headers); - header[0..8].copy_from_slice(&checksum.to_le_bytes()); - - blob.extend_from_slice(&header); - blob.extend_from_slice(legacy.payload); - blob.extend_from_slice(legacy.user_headers); - } - - let blob = blob.freeze(); - let mut header = SendMessages2Header::new( - namespace.partition_id() as u64, - origin_timestamp, - u64::try_from(COMMAND_HEADER_SIZE + blob.len()) - .map_err(|_| IggyError::InvalidCommand)?, - message_count, - ); - header.batch_checksum = calculate_batch_checksum(&header, &blob); - - Ok(Self { header, blob }) - } - pub fn encode_request( self, mut request_header: RequestHeader, @@ -590,9 +518,27 @@ pub fn encrypt_batch_request( SendMessages2Owned { header, blob }.encode_request(request_header) } +/// Whether the legacy transcode stamps a batch checksum onto its output. +/// +/// The recompute is a full-blob `XxHash3` pass, needed only when a reader +/// validates the transcoded batch before [`stamp_prepare_for_persistence`] +/// recomputes it: the encrypt ingest path re-decodes the canonicalized batch +/// (`encrypt_batch_request`'s validating decode, then the second `convert` its +/// output re-enters as the canonical-vs-legacy discriminator). The partition +/// ingest path has no such reader, so it skips the pass and the checksum stays +/// zero until stamp. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ChecksumMode { + /// Compute the batch checksum over the transcoded blob. + Compute, + /// Leave the batch checksum zero; `stamp_prepare_for_persistence` fills it. + Skip, +} + pub fn convert_request_message( namespace: IggyNamespace, message: Message<RequestHeader>, + checksum: ChecksumMode, ) -> Result<Message<RequestHeader>, IggyError> { let request_header = *message.header(); let total_size = request_header.size as usize; @@ -600,7 +546,119 @@ pub fn convert_request_message( if decode_batch_slice(body).is_ok() { return Ok(message); } - SendMessages2Owned::from_legacy_request(namespace, body)?.encode_request(request_header) + transcode_legacy_request(namespace, body, request_header, checksum) +} + +/// Transcode a legacy `SendMessages` request body directly into the canonical +/// `[RequestHeader][256B SendMessages2Header][blob]` form, writing each message +/// record straight into the final aligned buffer. +/// +/// Fused replacement for the `from_legacy_request(..).encode_request(..)` +/// two-step: a size walk over the legacy input sizes the single output +/// allocation, then a write walk lays down each canonical record in place. This +/// drops the intermediate blob allocation and the full-blob copy the two-step +/// paid. Output bytes are identical to that path. +/// +/// `checksum` selects whether the output carries a batch checksum (see +/// [`ChecksumMode`]); [`ChecksumMode::Skip`] leaves it zero for the partition +/// ingest path, where stamp recomputes it. +fn transcode_legacy_request( + namespace: IggyNamespace, + body: &[u8], + mut request_header: RequestHeader, + checksum: ChecksumMode, +) -> Result<Message<RequestHeader>, 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; + let mut cursor = 0usize; + let mut blob_len = 0usize; + + while cursor < messages.len() && parsed.len() < message_count as usize { + let legacy = LegacyMessageRef::decode(&messages[cursor..])?; + origin_timestamp = origin_timestamp.min(legacy.origin_timestamp); + cursor += legacy.total_size; + blob_len = blob_len + .checked_add(MESSAGE_HEADER_SIZE + legacy.payload.len() + legacy.user_headers.len()) + .ok_or(IggyError::InvalidCommand)?; + parsed.push(legacy); + } + + if parsed.len() != message_count as usize || cursor != messages.len() { + return Err(IggyError::InvalidCommand); + } + + if origin_timestamp == u64::MAX { + origin_timestamp = 0; + } + + let header_size = std::mem::size_of::<RequestHeader>(); + let batch_length = COMMAND_HEADER_SIZE + .checked_add(blob_len) + .ok_or(IggyError::InvalidCommand)?; + let total_size = header_size + .checked_add(batch_length) + .ok_or(IggyError::InvalidCommand)?; + 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..header_size].copy_from_slice(bytemuck::bytes_of(&request_header)); + + let mut write = PREPARE_SPLIT_POINT; + for (index, legacy) in parsed.iter().enumerate() { + let id = if legacy.id == 0 { + random_id::get_uuid() + } else { + legacy.id + }; + let offset_delta = u32::try_from(index).map_err(|_| IggyError::InvalidCommand)?; + let timestamp_delta = legacy + .origin_timestamp + .checked_sub(origin_timestamp) + .ok_or(IggyError::InvalidCommand)?; + if timestamp_delta > MAX_TIMESTAMP_DELTA_MICROS { + return Err(IggyError::InvalidMessageTimestampDelta(timestamp_delta)); + } + let timestamp_delta = + u32::try_from(timestamp_delta).map_err(|_| IggyError::InvalidCommand)?; + let user_headers_length = + u32::try_from(legacy.user_headers.len()).map_err(|_| IggyError::InvalidCommand)?; + let payload_length = + u32::try_from(legacy.payload.len()).map_err(|_| IggyError::InvalidCommand)?; + + let mut header = [0u8; MESSAGE_HEADER_SIZE]; + header[8..24].copy_from_slice(&id.to_le_bytes()); + header[24..28].copy_from_slice(&offset_delta.to_le_bytes()); + header[28..32].copy_from_slice(×tamp_delta.to_le_bytes()); + header[32..36].copy_from_slice(&user_headers_length.to_le_bytes()); + header[36..40].copy_from_slice(&payload_length.to_le_bytes()); + let checksum = calculate_checksum_parts(&header[8..], legacy.payload, legacy.user_headers); + header[0..8].copy_from_slice(&checksum.to_le_bytes()); + + bytes[write..write + MESSAGE_HEADER_SIZE].copy_from_slice(&header); + write += MESSAGE_HEADER_SIZE; + bytes[write..write + legacy.payload.len()].copy_from_slice(legacy.payload); + write += legacy.payload.len(); + bytes[write..write + legacy.user_headers.len()].copy_from_slice(legacy.user_headers); + write += legacy.user_headers.len(); + } + + let mut command = SendMessages2Header::new( + namespace.partition_id() as u64, + origin_timestamp, + batch_length as u64, + message_count, + ); + if checksum == ChecksumMode::Compute { + command.batch_checksum = calculate_batch_checksum( + &command, + &bytes[PREPARE_SPLIT_POINT..PREPARE_SPLIT_POINT + blob_len], + ); + } + command.encode_into(&mut bytes[header_size..header_size + COMMAND_HEADER_SIZE]); + + Message::try_from(buffer).map_err(|_| IggyError::InvalidCommand) } /// Decode one batch slice (`[256B command header][blob]`), validating the @@ -630,7 +688,8 @@ pub fn decode_batch_slice(body: &[u8]) -> Result<SendMessages2Ref<'_>, IggyError Ok(SendMessages2Ref { header, blob }) } -/// Decode a `Prepare` message from a slice of bytes. +/// Decode a `Prepare` message from a slice of bytes, validating the batch +/// checksum. /// /// `bytes` must be 16-byte aligned (`PrepareHeader` has `u128` fields). Source /// from `Frozen<MESSAGE_ALIGN>` / `Owned<MESSAGE_ALIGN>` / `Message<H>`. @@ -641,6 +700,37 @@ pub fn decode_batch_slice(body: &[u8]) -> Result<SendMessages2Ref<'_>, IggyError /// `IggyError::InvalidCommand` on: short buffer, bad bit pattern, `size` /// outside `[header_size, bytes.len()]`, short/checksum-mismatched body. pub fn decode_prepare_slice(bytes: &[u8]) -> Result<SendMessages2Ref<'_>, IggyError> { + decode_prepare_slice_inner(bytes, true) +} + +/// Like [`decode_prepare_slice`] but skips the full-blob batch-checksum +/// recomputation, extracting only the header meta. Every cheap structural check +/// (length, 16-byte alignment, `size` bounds, blob length) is still enforced. +/// +/// INVARIANT: `bytes` MUST be node-local self-stamped - +/// [`stamp_prepare_for_persistence`] recomputed the batch checksum over the +/// exact blob on THIS node - or already integrity-checked at their network +/// ingress. There is no consensus-layer blob validation: the `PrepareHeader` +/// integrity fields are inert zeros. A replicated `SendMessages` prepare is +/// gated per-message on receipt by [`verify_received_send_messages`], and a +/// repaired prepare is validated via [`decode_prepare_slice`]; both run BEFORE +/// the bytes reach any trusted decode. NEVER call this on unvalidated network +/// bytes - it would let a corrupted blob pass undetected. The `XxHash3` pass +/// over a ~1 MiB blob dominates produce-path CPU, so trusted call sites that +/// only read header meta skip it. +/// +/// # Errors +/// +/// Same structural errors as [`decode_prepare_slice`], minus +/// `InvalidBatchChecksum`. +pub fn decode_prepare_slice_trusted(bytes: &[u8]) -> Result<SendMessages2Ref<'_>, IggyError> { + decode_prepare_slice_inner(bytes, false) +} + +fn decode_prepare_slice_inner( + bytes: &[u8], + validate_checksum: bool, +) -> Result<SendMessages2Ref<'_>, IggyError> { let header_size = std::mem::size_of::<PrepareHeader>(); if bytes.len() < header_size { return Err(IggyError::InvalidCommand); @@ -677,13 +767,15 @@ pub fn decode_prepare_slice(bytes: &[u8]) -> Result<SendMessages2Ref<'_>, IggyEr } let blob = &blob[..blob_len]; - let expected_checksum = calculate_batch_checksum(&header, blob); - if header.batch_checksum != expected_checksum { - return Err(IggyError::InvalidBatchChecksum( - header.batch_checksum, - expected_checksum, - header.base_offset, - )); + if validate_checksum { + let expected_checksum = calculate_batch_checksum(&header, blob); + if header.batch_checksum != expected_checksum { + return Err(IggyError::InvalidBatchChecksum( + header.batch_checksum, + expected_checksum, + header.base_offset, + )); + } } Ok(SendMessages2Ref { header, blob }) @@ -711,6 +803,57 @@ pub fn stamp_prepare_for_persistence( Ok((message, command, command.message_count)) } +/// Verify every per-message checksum in a received `SendMessages` prepare. +/// +/// The FIRST blob-integrity check on the replicated path: the `PrepareHeader` +/// integrity fields are inert zeros and the batch checksum is recomputed +/// locally at stamp, so transit corruption of a message body would otherwise +/// reach apply undetected. Backups call this before journaling a replicated +/// prepare; on a mismatch the caller fails closed (drop, no `PrepareOk`) and the +/// primary retransmits on prepare-timeout. +/// +/// The decode is structural only (no batch-checksum recompute). Each message's +/// stored checksum is recomputed over its stamp-invariant cover +/// (`header[8..48] || payload || user_headers`), which excludes the 256B command +/// header, so it holds whether or not this node has stamped `base_offset` / +/// `base_timestamp` yet (a received prepare is pre-stamp). +/// +/// # Errors +/// +/// [`IggyError::InvalidCommand`] if the records do not tile `message_count` +/// exactly (a length-field corruption desyncs the walk); +/// [`IggyError::InvalidMessageChecksum`] on the first per-message mismatch. +pub fn verify_received_send_messages(bytes: &[u8]) -> Result<(), IggyError> { + let batch = decode_prepare_slice_trusted(bytes)?; + let blob = batch.blob(); + let mut verified = 0u32; + let mut covered = 0usize; + for framed in batch.iter_with_offsets() { + // Raw header tail (`header[8..48]`, the checksum-covered fields plus the + // zero reserved bytes) sourced from the blob, not rebuilt from decoded + // fields, so the cover is byte-exact with the encoder's. + let header_tail = &blob[framed.start + 8..framed.start + MESSAGE_HEADER_SIZE]; + let expected = calculate_checksum_parts( + header_tail, + framed.message.payload, + framed.message.user_headers, + ); + if expected != framed.message.header.checksum { + return Err(IggyError::InvalidMessageChecksum( + framed.message.header.checksum, + expected, + batch.header.base_offset + u64::from(framed.message.header.offset_delta), + )); + } + verified += 1; + covered = framed.end; + } + if verified != batch.message_count() || covered != blob.len() { + return Err(IggyError::InvalidCommand); + } + Ok(()) +} + fn legacy_messages_slice(body: &[u8]) -> Result<(u32, &[u8]), IggyError> { if body.len() < 4 { return Err(IggyError::InvalidCommand); @@ -827,7 +970,8 @@ fn read_u128(bytes: &[u8], offset: usize) -> Result<u128, IggyError> { #[cfg(test)] mod tests { use super::*; - use iggy_binary_protocol::Command2; + use iggy_binary_protocol::{Command2, Operation}; + use iggy_common::Aes256GcmEncryptor; fn aligned_prepare_bytes(size: u32) -> Owned<MESSAGE_ALIGN> { let mut owned = Owned::<MESSAGE_ALIGN>::zeroed(std::mem::size_of::<PrepareHeader>()); @@ -839,6 +983,82 @@ mod tests { owned } + /// A checksum-consistent `Prepare`: `[PrepareHeader][256B batch header][blob]` + /// with `batch_checksum` stamped over the final header fields + `blob`. + fn valid_prepare_bytes(blob: &[u8]) -> Owned<MESSAGE_ALIGN> { + let header_size = std::mem::size_of::<PrepareHeader>(); + let batch_length = COMMAND_HEADER_SIZE + blob.len(); + let total = header_size + batch_length; + let mut owned = Owned::<MESSAGE_ALIGN>::zeroed(total); + { + let prepare: &mut PrepareHeader = + bytemuck::checked::try_from_bytes_mut(&mut owned.as_mut_slice()[..header_size]) + .expect("zeroed bytes form a valid PrepareHeader"); + prepare.command = Command2::Prepare; + prepare.size = u32::try_from(total).expect("prepare size fits u32"); + } + + let mut command = SendMessages2Header::new(7, 123, batch_length as u64, 3); + command.base_offset = 10; + command.base_timestamp = 20; + command.batch_checksum = command.checksum_for_blob(blob); + + let bytes = owned.as_mut_slice(); + command.encode_into(&mut bytes[header_size..header_size + COMMAND_HEADER_SIZE]); + bytes[header_size + COMMAND_HEADER_SIZE..].copy_from_slice(blob); + owned + } + + #[test] + fn decode_prepare_slice_trusted_matches_validating_for_valid_batch() { + // The trusted variant must surface byte-identical header meta to the + // validating decode for a checksum-consistent batch; only the full-blob + // hash pass is skipped. + let blob = vec![0x5Au8; 4096]; + let owned = valid_prepare_bytes(&blob); + + let validated = decode_prepare_slice(owned.as_slice()).expect("valid batch decodes"); + let trusted = + decode_prepare_slice_trusted(owned.as_slice()).expect("valid batch decodes trusted"); + + assert_eq!(validated.header.base_offset, trusted.header.base_offset); + assert_eq!( + validated.header.base_timestamp, + trusted.header.base_timestamp + ); + assert_eq!( + validated.header.origin_timestamp, + trusted.header.origin_timestamp + ); + assert_eq!(validated.header.batch_length, trusted.header.batch_length); + assert_eq!(validated.message_count(), trusted.message_count()); + assert_eq!(validated.header.total_size(), trusted.header.total_size()); + assert_eq!(validated.blob(), trusted.blob()); + } + + #[test] + fn decode_prepare_slice_trusted_skips_batch_checksum() { + // A blob mutated after stamping fails the validating decode but passes + // the trusted one: exactly why the trusted variant is confined to + // locally-produced bytes (see its doc invariant). + let blob = vec![0x11u8; 512]; + let mut owned = valid_prepare_bytes(&blob); + let corrupt_index = owned.as_slice().len() - 1; + owned.as_mut_slice()[corrupt_index] ^= 0xFF; + + assert!( + matches!( + decode_prepare_slice(owned.as_slice()), + Err(IggyError::InvalidBatchChecksum(..)) + ), + "validating decode must reject a corrupted blob", + ); + assert!( + decode_prepare_slice_trusted(owned.as_slice()).is_ok(), + "trusted decode skips the batch-checksum recomputation", + ); + } + #[test] fn decode_prepare_slice_size_below_header_size_does_not_panic() { // Regression: without the `total_size < header_size` guard, @@ -869,4 +1089,246 @@ mod tests { ); let _ = decode_prepare_slice(misaligned); } + + fn sample_messages() -> IggyMessages2 { + let mut messages = IggyMessages2::with_capacity(2); + messages.push(IggyMessage2 { + header: IggyMessage2Header { + id: 7, + origin_timestamp: 1_000, + ..Default::default() + }, + payload: Bytes::from_static(b"first-payload"), + user_headers: None, + }); + messages.push(IggyMessage2 { + header: IggyMessage2Header { + id: 8, + origin_timestamp: 1_050, + ..Default::default() + }, + payload: Bytes::from_static(b"second-payload"), + user_headers: Some(Bytes::from_static(b"user-header-bytes")), + }); + messages + } + + /// `[PrepareHeader][256B batch header][blob]` carrying real per-message + /// records + checksums from the production encoder, left pre-stamp + /// (`base_offset` / `base_timestamp` zero) as a follower receives it. + fn prepare_with_messages(messages: &IggyMessages2) -> Owned<MESSAGE_ALIGN> { + let namespace = IggyNamespace::new(1, 1, 7); + let owned = + SendMessages2Owned::from_messages(namespace, messages).expect("build send batch"); + let header_size = std::mem::size_of::<PrepareHeader>(); + let total = header_size + owned.header.total_size(); + let mut buffer = Owned::<MESSAGE_ALIGN>::zeroed(total); + { + let prepare: &mut PrepareHeader = + bytemuck::checked::try_from_bytes_mut(&mut buffer.as_mut_slice()[..header_size]) + .expect("zeroed bytes form a valid PrepareHeader"); + prepare.command = Command2::Prepare; + prepare.size = u32::try_from(total).expect("prepare size fits u32"); + } + let bytes = buffer.as_mut_slice(); + owned + .header + .encode_into(&mut bytes[header_size..header_size + COMMAND_HEADER_SIZE]); + bytes[PREPARE_SPLIT_POINT..PREPARE_SPLIT_POINT + owned.blob.len()] + .copy_from_slice(&owned.blob); + buffer + } + + #[test] + fn verify_received_send_messages_accepts_clean_batch() { + let owned = prepare_with_messages(&sample_messages()); + verify_received_send_messages(owned.as_slice()) + .expect("a clean batch passes the receive gate"); + } + + #[test] + fn verify_received_send_messages_rejects_flipped_payload_byte() { + let mut owned = prepare_with_messages(&sample_messages()); + // First payload begins right after the first message's 48B header. + let payload_index = PREPARE_SPLIT_POINT + MESSAGE_HEADER_SIZE; + owned.as_mut_slice()[payload_index] ^= 0xFF; + assert!( + matches!( + verify_received_send_messages(owned.as_slice()), + Err(IggyError::InvalidMessageChecksum(..)) + ), + "a flipped payload byte must fail the per-message checksum", + ); + } + + #[test] + fn verify_received_send_messages_rejects_flipped_stored_checksum() { + let mut owned = prepare_with_messages(&sample_messages()); + // The first message's stored checksum is the first 8 bytes of the blob. + owned.as_mut_slice()[PREPARE_SPLIT_POINT] ^= 0xFF; + assert!( + matches!( + verify_received_send_messages(owned.as_slice()), + Err(IggyError::InvalidMessageChecksum(..)) + ), + "a flipped stored checksum must fail the per-message check", + ); + } + + /// Legacy `SendMessages` request body: `[metadata_len=4][message_count]` + /// then `count` skipped index slots, then the 64B-header legacy records. + fn legacy_send_messages_body(messages: &IggyMessages2) -> Vec<u8> { + let count = messages.count(); + let mut body = Vec::new(); + body.extend_from_slice(&4u32.to_le_bytes()); + body.extend_from_slice(&count.to_le_bytes()); + body.extend_from_slice(&vec![0u8; count as usize * INDEX_SIZE]); + for message in messages.iter() { + let user_headers = message.user_headers.as_deref().unwrap_or_default(); + let mut header = [0u8; LEGACY_MESSAGE_HEADER_SIZE]; + header[8..24].copy_from_slice(&message.header.id.to_le_bytes()); + header[40..48].copy_from_slice(&message.header.origin_timestamp.to_le_bytes()); + header[48..52].copy_from_slice(&(user_headers.len() as u32).to_le_bytes()); + header[52..56].copy_from_slice(&(message.payload.len() as u32).to_le_bytes()); + body.extend_from_slice(&header); + body.extend_from_slice(&message.payload); + body.extend_from_slice(user_headers); + } + body + } + + fn legacy_request_message(body: &[u8]) -> Message<RequestHeader> { + let header_size = std::mem::size_of::<RequestHeader>(); + let total = header_size + body.len(); + let mut buffer = Owned::<MESSAGE_ALIGN>::zeroed(total); + { + let header: &mut RequestHeader = + bytemuck::checked::try_from_bytes_mut(&mut buffer.as_mut_slice()[..header_size]) + .expect("zeroed bytes form a valid RequestHeader"); + header.command = Command2::Request; + header.operation = Operation::SendMessages; + header.client = 1; + header.session = 1; + header.request = 1; + header.size = u32::try_from(total).expect("size fits u32"); + } + buffer.as_mut_slice()[header_size..].copy_from_slice(body); + Message::try_from(buffer).expect("legacy request message is valid") + } + + #[test] + fn convert_request_message_transcodes_legacy_to_canonical_bytes() { + // Golden: the fused legacy transcode must emit the exact canonical batch + // the native builder (`from_messages`) produces for the same messages - + // command header + blob, byte for byte. Explicit non-zero ids keep it + // deterministic (no `random_id` substitution). + let namespace = IggyNamespace::new(1, 1, 3); + let messages = sample_messages(); + + let owned = + SendMessages2Owned::from_messages(namespace, &messages).expect("build canonical batch"); + let mut expected_body = vec![0u8; COMMAND_HEADER_SIZE + owned.blob.len()]; + owned + .header + .encode_into(&mut expected_body[..COMMAND_HEADER_SIZE]); + expected_body[COMMAND_HEADER_SIZE..].copy_from_slice(&owned.blob); + + 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 actual_body = &converted.as_slice()[header_size..converted.header().size as usize]; + + assert_eq!( + actual_body, expected_body, + "legacy transcode must be byte-identical to the canonical native batch", + ); + + // And the emitted batch is self-consistent: it validates through the + // batch-checksum decode and yields the original messages. + let decoded = decode_batch_slice(actual_body).expect("transcoded batch checksum is valid"); + assert_eq!(decoded.message_count(), messages.count()); + let payloads: Vec<&[u8]> = decoded.iter().map(|view| view.payload).collect(); + assert_eq!( + payloads, + vec![&b"first-payload"[..], &b"second-payload"[..]] + ); + } + + #[test] + fn convert_request_message_skip_leaves_batch_checksum_zero_until_stamp() { + // The partition ingest path passes Skip: the transcoded batch must carry + // a zero checksum (stamp fills it) and be otherwise byte-identical to the + // Compute output - the flag toggles nothing but that one hash. + 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 computed = convert_request_message( + namespace, + legacy_request_message(&body), + ChecksumMode::Compute, + ) + .expect("compute transcode"); + let skipped = + convert_request_message(namespace, legacy_request_message(&body), ChecksumMode::Skip) + .expect("skip transcode"); + + let computed_body = &computed.as_slice()[header_size..computed.header().size as usize]; + let skipped_body = &skipped.as_slice()[header_size..skipped.header().size as usize]; + + let skipped_header = SendMessages2Header::decode(&skipped_body[..COMMAND_HEADER_SIZE]) + .expect("decode skipped header"); + assert_eq!( + skipped_header.batch_checksum, 0, + "skip leaves the batch checksum zero until stamp", + ); + + // Patch only the 8-byte batch_checksum field into the skipped body; it + // must then equal the computed body, proving nothing else diverges. + let mut patched = skipped_body.to_vec(); + patched[BATCH_CHECKSUM_OFFSET..BATCH_CHECKSUM_OFFSET + 8] + .copy_from_slice(&computed_body[BATCH_CHECKSUM_OFFSET..BATCH_CHECKSUM_OFFSET + 8]); + assert_eq!( + patched.as_slice(), + computed_body, + "skip and compute differ only in the batch_checksum field", + ); + } + + #[test] + fn encrypt_ingest_path_stays_canonical_through_flag_split() { + // Mirror the plane encrypt ingest sequence: convert(Compute) -> the + // validating decode encrypt performs on its input -> encrypt -> the + // validating decode the second convert performs as its discriminator -> + // convert(Skip) (the partition convert), which sees an already-canonical + // 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 legacy = legacy_request_message(&legacy_send_messages_body(&messages)); + let canonical = convert_request_message(namespace, legacy, ChecksumMode::Compute) + .expect("pre-encrypt transcode"); + let canonical_body = &canonical.as_slice()[header_size..canonical.header().size as usize]; + decode_batch_slice(canonical_body).expect("encrypt input decode validates the checksum"); + + let encryptor = + EncryptorKind::Aes256Gcm(Aes256GcmEncryptor::new(&[7u8; 32]).expect("valid 32B key")); + let encrypted = encrypt_batch_request(canonical, &encryptor).expect("encrypt batch"); + let encrypted_body: Vec<u8> = + encrypted.as_slice()[header_size..encrypted.header().size as usize].to_vec(); + decode_batch_slice(&encrypted_body) + .expect("encrypt output drives the 2nd-convert discriminator"); + + let repassed = convert_request_message(namespace, encrypted, ChecksumMode::Skip) + .expect("second convert passes the canonical batch"); + let repassed_body = &repassed.as_slice()[header_size..repassed.header().size as usize]; + assert_eq!( + repassed_body, + encrypted_body.as_slice(), + "an already-canonical encrypted batch passes the partition convert untouched", + ); + } } diff --git a/core/simulator/src/client.rs b/core/simulator/src/client.rs index f511c1000..028adb38c 100644 --- a/core/simulator/src/client.rs +++ b/core/simulator/src/client.rs @@ -72,7 +72,7 @@ pub struct SimClient { partition_counter: Cell<u64>, /// Deterministic per-message id source for produced messages. The real SDK /// sends `id: 0` and lets the server mint a random UUID - /// (`SendMessages2::from_legacy_request` -> `random_id::get_uuid`); that + /// (`transcode_legacy_request` -> `random_id::get_uuid`); that /// mint is unseeded, so under the deterministic executor a produce's /// replicated body bytes (and their checksums) would differ run to run, /// silently breaking seeded replay. Stamping a deterministic id here keeps @@ -516,7 +516,7 @@ impl SimClient { /// client). VSR clients resolve to an explicit partition before sending, so /// the sim always emits `WirePartitioning::PartitionId`: that is the shape /// the shell's `resolve_partition_request_namespace` decodes, and the raw - /// path converts it to `SendMessages2` via `from_legacy_request`. + /// path converts it to `SendMessages2` via `transcode_legacy_request`. /// /// # Panics /// Panics if a namespace id exceeds `u32` or the request buffer is invalid.
