hubcio commented on code in PR #3487:
URL: https://github.com/apache/iggy/pull/3487#discussion_r3421279548
##########
core/shard/src/lib.rs:
##########
@@ -985,6 +1156,54 @@ where
}
}
+ /// Park a partition-plane frame whose namespace this shard has not yet
+ /// materialised (post-`CreateTopic` convergence window: the metadata
+ /// commit precedes the reconciler pass that builds the local replica).
+ ///
+ /// Returns `Some(message)` when the frame should be processed normally:
+ /// non-partition operation, namespace materialised, or namespace
+ /// tombstoned (the plane's own tombstone guard handles the drop).
+ /// Returns `None` when the frame was parked (or dropped on overflow);
+ /// [`Self::apply_reconcile_ops`] re-dispatches parked frames once the
+ /// matching `ReconcileOp::InsertOwned` lands.
+ fn park_if_unmaterialised<H>(
Review Comment:
parked frames are only ever drained on `ReconcileOp::InsertOwned`. if a
topic is created (frames parked) then deleted before the reconciler
materialises it, the namespace never reaches `InsertOwned`, so
`pending_partition_frames[ns]` is never reclaimed - it leaks until process
exit. bounded at 128/ns but unbounded across many create-delete-raced
namespaces. drop the pending entry on tombstone / remove-routed /
confirm-remove too.
##########
core/server-ng/src/responses.rs:
##########
@@ -744,3 +746,121 @@ pub(crate) fn current_metadata_commit(shard:
&Rc<ServerNgShard>) -> u64 {
.as_ref()
.map_or(0, VsrConsensus::commit_max)
}
+
+/// 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`.
+const STORED_MESSAGE_HEADER_SIZE: usize = 48;
+
+/// Build the `PolledMessages` reply body from the owning shard's poll
+/// fragments.
+///
+/// Fragments carry the stored `SendMessages2` batches: a 256-byte command
+/// header followed by `IggyMessage2`-format messages
+/// (`[48B header][payload][user_headers]`, offsets/timestamps delta-encoded
+/// against the batch). The SDK decodes the legacy wire format
+/// (`[64B header][payload][user_headers]`, absolute offsets); the message
+/// sections share the legacy order, so only the header is re-encoded here
+/// and the section bytes copy through contiguously.
+///
+/// Body layout: `[partition_id:4][current_offset:8][count:4][messages...]`.
+pub(crate) fn build_polled_messages_body(
+ partition_id: u32,
+ current_offset: u64,
+ fragments: PollFragments,
+) -> Result<Bytes, IggyError> {
+ // Batches may arrive split across fragments (rewritten command header +
+ // sliced blob); concatenate into one stream before walking batches.
+ let mut stream: Vec<u8> = Vec::new();
+ for fragment in fragments {
+ let frozen = fragment.into_frozen();
+ stream.extend_from_slice(frozen.as_slice());
+ }
+
+ let mut messages: Vec<u8> = Vec::with_capacity(stream.len());
+ let mut count: u32 = 0;
+ let mut position = 0usize;
+ while position < stream.len() {
+ let batch = SendMessages2Header::decode(&stream[position..])?;
+ let batch_end = position
+ .checked_add(
+ usize::try_from(batch.batch_length).map_err(|_|
IggyError::InvalidCommand)?,
+ )
+ .ok_or(IggyError::InvalidCommand)?;
+ if batch_end > stream.len() {
+ return Err(IggyError::InvalidCommand);
+ }
+ let mut cursor = position + COMMAND_HEADER_SIZE;
+ while cursor < batch_end {
+ if cursor + STORED_MESSAGE_HEADER_SIZE > batch_end {
+ return Err(IggyError::InvalidCommand);
+ }
+ let header = &stream[cursor..cursor + STORED_MESSAGE_HEADER_SIZE];
+ let checksum = &header[0..8];
+ let id = &header[8..24];
+ let offset_delta =
u32::from_le_bytes(header[24..28].try_into().expect("4-byte slice"));
+ let timestamp_delta =
+ u32::from_le_bytes(header[28..32].try_into().expect("4-byte
slice"));
+ let user_headers_length =
+ u32::from_le_bytes(header[32..36].try_into().expect("4-byte
slice")) as usize;
+ let payload_length =
+ u32::from_le_bytes(header[36..40].try_into().expect("4-byte
slice")) as usize;
+
+ let sections_start = cursor + STORED_MESSAGE_HEADER_SIZE;
+ let sections_end = sections_start + payload_length +
user_headers_length;
+ if sections_end > batch_end {
+ return Err(IggyError::InvalidCommand);
+ }
+
+ let offset = batch.base_offset + u64::from(offset_delta);
+ let timestamp = batch.base_timestamp + u64::from(timestamp_delta);
Review Comment:
this mixes two clocks. `base_timestamp` is the broker wall-clock stamped
once per batch (`IggyTimestamp::now()` in `iggy_partition` before persist), but
`timestamp_delta` is a per-message delta against the producer origin timestamp,
not the broker base. so `base_timestamp + timestamp_delta` gives a meaningless
per-message `timestamp` for any batch with more than one distinct origin
timestamp. the `origin_timestamp` recon on the next line is correct. fix: `let
timestamp = batch.base_timestamp;` (drop the delta) - that's the flat broker
append time, same as the currently-dead `owned_message` reader. the round-trip
test passes only because its messages share an origin timestamp (delta = 0) and
it never asserts the timestamp field.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -674,6 +687,269 @@ where
IggyNamespace::from_raw(self.consensus.namespace())
}
+ /// Resolve a poll query against the in-memory journal, falling back to
+ /// the on-disk segments for ranges the journal no longer holds (the
+ /// persist threshold drains committed batches to segment files).
+ ///
+ /// A query is served from exactly one tier per call: a poll that starts
+ /// below the journal's oldest resident offset reads from disk only, and
+ /// the client's next poll (advancing past what was returned) eventually
+ /// crosses back into the resident range. Timestamp queries try disk
+ /// first whenever segments hold persisted bytes -- older matches always
+ /// live there -- and fall back to the journal when the disk has none.
+ async fn lookup_messages(&self, query: MessageLookup) ->
Option<PollQueryResult<4096>> {
+ let serve_journal_first = match query {
+ MessageLookup::Offset { offset, .. } => self
+ .log
+ .journal()
+ .inner
+ .oldest_resident_offset()
+ .is_some_and(|oldest| offset >= oldest),
+ MessageLookup::Timestamp { .. } =>
!self.has_persisted_segment_bytes(),
+ };
+
+ if serve_journal_first {
+ return self.log.journal().inner.get(&query).await;
+ }
+ match self.poll_from_disk(query).await {
+ Some((mut fragments, last_matching_offset, matched)) => {
+ // A poll can straddle the tiers: older messages already
+ // drained to segments, the tail still journal-resident.
+ // Continue past the last disk match by offset (timestamp
+ // matches are contiguous from the first hit, so an offset
+ // continuation is equivalent).
+ let remaining = query.count().saturating_sub(matched);
+ if remaining > 0
+ && let Some(last_offset) = last_matching_offset
+ {
+ let continuation = MessageLookup::Offset {
+ offset: last_offset + 1,
+ count: remaining,
+ };
+ if let Some((journal_fragments, journal_last)) =
+ self.log.journal().inner.get(&continuation).await
+ {
+ fragments.extend(journal_fragments);
+ return Some((fragments,
journal_last.or(last_matching_offset)));
+ }
+ }
+ Some((fragments, last_matching_offset))
+ }
+ // Nothing matched on disk (e.g. a timestamp newer than every
+ // persisted batch): the match, if any, is journal-resident.
+ None => self.log.journal().inner.get(&query).await,
+ }
+ }
+
+ fn partition_dir(&self) -> Option<String> {
+ if self.partition_dir.is_some() {
+ return self.partition_dir.clone();
+ }
+ // Writer-derived fallback for partitions built without
+ // `set_partition_dir`. Unreliable mid-rotation: sealed segments
+ // drop their writer, so prefer the stored path above.
+ self.log
+ .messages_writers()
+ .iter()
+ .rev()
+ .flatten()
+ .next()
+ .and_then(|writer| {
+ std::path::Path::new(&writer.path())
+ .parent()
+ .map(|dir| dir.to_string_lossy().into_owned())
+ })
+ }
+
+ fn has_persisted_segment_bytes(&self) -> bool {
+ self.log
+ .segments()
+ .iter()
+ .any(|segment| segment.size.as_bytes_u64() > 0)
+ }
+
+ /// Serve a poll from the on-disk segment files.
+ ///
+ /// Picks the starting segment + byte position via the sparse index
+ /// (one entry per persist flush; a miss falls back to the segment
+ /// start), then walks stamped `[256B SendMessages2Header][blob]`
+ /// batches in chunked reads, slicing fragments with the same selector
+ /// the journal path uses. Batches split across a chunk boundary are
+ /// re-read from their start in the next chunk.
+ #[allow(clippy::cast_possible_truncation)]
+ async fn poll_from_disk(
+ &self,
+ query: MessageLookup,
+ ) -> Option<(PollFragments<4096>, Option<u64>, u32)> {
+ const DISK_POLL_CHUNK: u64 = 1 << 20;
+
+ let count = query.count();
+ if count == 0 || !self.log.has_segments() {
+ return None;
+ }
+
+ let (start_segment, mut position) = self.disk_poll_start(&query);
+
+ let mut fragments = PollFragments::new();
+ let mut last_matching_offset = None;
+ let mut matched: u32 = 0;
+
+ for segment_index in start_segment..self.log.segments().len() {
+ if matched >= count {
+ break;
+ }
+ let persisted =
self.log.segments()[segment_index].size.as_bytes_u64();
+ if persisted == 0 || position >= persisted {
+ position = 0;
+ continue;
+ }
+ // Sealed segments drop their writer at rotation, so resolve the
+ // file from the partition directory (taken from any live writer)
+ // plus the segment's start offset, mirroring the writer naming.
+ let Some(partition_dir) = self.partition_dir() else {
+ // Simulated in-memory persistence: no files to read. A live
+ // partition hitting this means no writer was resolvable
+ // (e.g. mid-rotation), which silently hides the disk tier.
+ warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = self.namespace().inner(),
+ segment_count = self.log.segments().len(),
+ "disk poll: no live writer to resolve partition dir; disk
tier unreadable"
+ );
+ return None;
+ };
+ let start_offset = self.log.segments()[segment_index].start_offset;
+ let path = format!("{partition_dir}/{start_offset:0>20}.log");
+ let Some(file) = self.open_segment_with_retry(&path).await else {
+ position = 0;
+ continue;
+ };
+
+ let mut chunk_len = DISK_POLL_CHUNK;
+ while matched < count && position < persisted {
+ let len = (persisted - position).min(chunk_len) as usize;
+ let Some(chunk) = self.read_chunk_with_retry(&file, position,
len).await else {
+ break;
+ };
+ let consumed = walk_disk_chunk(
+ &chunk,
+ query,
+ count,
+ &mut matched,
+ &mut fragments,
+ &mut last_matching_offset,
+ );
+ if consumed == 0 {
+ if (len as u64) >= persisted - position {
+ // The whole remainder fit and still no complete
+ // batch decoded: corrupt tail; stop.
+ break;
+ }
+ // A single batch larger than the chunk: grow and
+ // re-read instead of spinning.
+ chunk_len = chunk_len.saturating_mul(4);
+ continue;
+ }
+ chunk_len = DISK_POLL_CHUNK;
+ position += consumed as u64;
+ }
+ position = 0;
+ }
+
+ if fragments.is_empty() {
+ None
+ } else {
+ Some((fragments, last_matching_offset, matched))
+ }
+ }
+
+ /// 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.
+ async fn open_segment_with_retry(&self, path: &str) ->
Option<compio::fs::File> {
+ for attempt in 0..3u8 {
+ match compio::fs::File::open(path).await {
+ Ok(file) => return Some(file),
+ Err(error) => {
+ warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = self.namespace().inner(),
+ path,
+ attempt,
+ %error,
+ "disk poll: failed to open segment file"
+ );
+
compio::time::sleep(std::time::Duration::from_millis(10)).await;
+ }
+ }
+ }
+ None
+ }
+
+ /// Read one chunk for a disk poll, retrying transient IO failures.
+ async fn read_chunk_with_retry(
+ &self,
+ file: &compio::fs::File,
+ position: u64,
+ len: usize,
+ ) -> Option<Frozen<4096>> {
+ for attempt in 0..3u8 {
+ let buffer = Owned::<4096>::zeroed(len);
Review Comment:
`Owned::<4096>::zeroed(len)` zeroes up to the full `DISK_POLL_CHUNK` (1 MiB)
and then `read_exact_at` immediately overwrites all of it - the memset is
wasted since the read fills the whole range before anything reads it. allocate
uninitialised, or reuse a single chunk buffer across the segment loop instead
of a fresh zeroed one per chunk.
##########
core/server-ng/src/dispatch.rs:
##########
@@ -606,6 +763,293 @@ async fn send_non_replicated_bytes(
}
}
+/// Reject a pre-auth request with a typed `Eviction(NoSession)` frame.
+///
+/// The SDK's reply decoder maps eviction reasons to typed errors
+/// (`NoSession` -> `Unauthenticated`), so clients fail fast with the same
+/// error the legacy server returns instead of a body-decode failure. The
+/// eviction context is best-effort off the metadata consensus (peer shards
+/// have none; zeroes are cosmetic -- the SDK only reads the reason).
+#[allow(clippy::future_not_send)]
+async fn send_unauthenticated_eviction(shard: &Rc<ServerNgShard>,
transport_client_id: u128) {
+ let ctx = shard.plane.metadata().consensus.as_ref().map_or(
+ consensus::EvictionContext {
+ cluster: 0,
+ view: 0,
+ replica: 0,
+ },
+ consensus::EvictionContext::from_consensus,
+ );
+ let eviction = consensus::build_eviction_message(
+ ctx,
+ transport_client_id,
+ iggy_binary_protocol::EvictionReason::NoSession,
+ );
+ if let Err(error) = shard
+ .bus
+ .send_to_client(transport_client_id,
eviction.into_generic().into_frozen())
+ .await
+ {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "failed to send unauthenticated eviction"
+ );
+ }
+}
+
+/// Serve `poll_messages`: resolve the partition namespace, run the read on
+/// the owning shard ([`shard::IggyShard::partition_read`]), and re-encode
+/// the stored batches into the legacy wire `PolledMessages` body.
+///
+/// Failures reply with an empty body so the SDK fails fast on decode
+/// instead of hanging until its read timeout.
+#[allow(clippy::future_not_send)]
+async fn handle_poll_messages(
+ shard: &Rc<ServerNgShard>,
+ transport_client_id: u128,
+ request: &Message<RequestHeader>,
+) {
+ let body = match decode_poll_request(shard, request) {
+ Ok((namespace, partition_id, consumer, args)) => {
+ match shard
+ .partition_read(namespace, PartitionRead::Poll { consumer,
args })
+ .await
+ {
+ Some(PartitionReadReply::Poll {
+ fragments,
+ current_offset,
+ }) => build_polled_messages_body(partition_id, current_offset,
fragments)
+ .unwrap_or_else(|error| {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "failed to re-encode polled batches; replying
empty poll"
+ );
+ empty_polled_messages_body(partition_id)
+ }),
+ other => {
+ warn!(
+ transport_client_id,
+ namespace = namespace.inner(),
+ reply_was_none = other.is_none(),
+ "partition read failed; replying empty poll"
+ );
+ empty_polled_messages_body(partition_id)
+ }
+ }
+ }
+ Err(error) => {
+ // A zero-byte body would panic the SDK's `PolledMessages`
+ // decoder; reply the 16-byte empty-poll shape instead.
+ warn!(
+ transport_client_id,
+ error = %error,
+ "poll_messages request rejected; replying empty poll"
+ );
+ empty_polled_messages_body(0)
+ }
+ };
+ send_non_replicated_bytes(shard, request, transport_client_id, body,
"poll_messages").await;
+}
+
+/// Serve `get_consumer_offset`. An empty body decodes as `None` on the SDK
+/// side (no offset stored / partition unknown).
+#[allow(clippy::future_not_send)]
+async fn handle_get_consumer_offset(
+ shard: &Rc<ServerNgShard>,
+ transport_client_id: u128,
+ request: &Message<RequestHeader>,
+) {
+ let body = match decode_consumer_offset_request(shard, request) {
+ Ok((namespace, partition_id, consumer)) => {
+ match shard
+ .partition_read(namespace, PartitionRead::ConsumerOffset {
consumer })
+ .await
+ {
+ Some(PartitionReadReply::ConsumerOffset {
+ stored: Some(stored_offset),
+ current_offset,
+ }) => build_consumer_offset_body(partition_id, current_offset,
stored_offset),
+ _ => Bytes::new(),
+ }
+ }
+ Err(error) => {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "get_consumer_offset request rejected; replying empty"
+ );
+ Bytes::new()
+ }
+ };
+ send_non_replicated_bytes(
+ shard,
+ request,
+ transport_client_id,
+ body,
+ "get_consumer_offset",
+ )
+ .await;
+}
+
+/// Wait (bounded) until `namespace` is routable: this shard's routing row
+/// exists and the owning shard answers a probe read (partition
+/// materialised). Fast path: row already present -> no probe, no wait.
+///
+/// Covers the post-`CreateTopic` convergence window where the metadata
+/// commit has returned to the client but the per-shard reconcilers have
+/// not yet seeded routing rows / materialised partitions.
+#[allow(clippy::future_not_send)]
+/// Ack a partition op that cannot be routed (unresolved or never-
+/// materialised namespace) with an empty Reply. The SDK connection
+/// processes replies in lockstep, so a silent drop wedges every
+/// subsequent request on that connection.
+async fn send_empty_partition_reply(
Review Comment:
the `Wait (bounded)...` doc block and the
`#[allow(clippy::future_not_send)]` here actually describe
`wait_for_partition_routable` below, not `send_empty_partition_reply`. the doc
is attached to the wrong fn and the allow is off-target - harmless today since
the lint isn't denied, but it won't suppress anything on the fn it was meant
for. move both down onto `wait_for_partition_routable`.
##########
core/server_common/src/send_messages2.rs:
##########
@@ -560,6 +564,12 @@ pub fn convert_request_message(
SendMessages2Owned::from_legacy_request(namespace,
body)?.encode_request(request_header)
}
+/// Decode one stored batch slice (`[256B command header][blob]`) -- the
+/// persisted segment-file record format. Validates the batch checksum.
+pub fn decode_batch_slice(body: &[u8]) -> Result<SendMessages2Ref<'_>,
IggyError> {
Review Comment:
`decode_batch_slice` is a verbatim one-line alias of the private
`decode_request_slice` with the same signature. either inline it at the single
caller or just make `decode_request_slice` pub - the persisted segment record
is the same layout as the request slice, so the extra name doesn't earn its
keep.
##########
core/server-ng/src/dispatch.rs:
##########
@@ -606,6 +763,293 @@ async fn send_non_replicated_bytes(
}
}
+/// Reject a pre-auth request with a typed `Eviction(NoSession)` frame.
+///
+/// The SDK's reply decoder maps eviction reasons to typed errors
+/// (`NoSession` -> `Unauthenticated`), so clients fail fast with the same
+/// error the legacy server returns instead of a body-decode failure. The
+/// eviction context is best-effort off the metadata consensus (peer shards
+/// have none; zeroes are cosmetic -- the SDK only reads the reason).
+#[allow(clippy::future_not_send)]
+async fn send_unauthenticated_eviction(shard: &Rc<ServerNgShard>,
transport_client_id: u128) {
+ let ctx = shard.plane.metadata().consensus.as_ref().map_or(
+ consensus::EvictionContext {
+ cluster: 0,
+ view: 0,
+ replica: 0,
+ },
+ consensus::EvictionContext::from_consensus,
+ );
+ let eviction = consensus::build_eviction_message(
+ ctx,
+ transport_client_id,
+ iggy_binary_protocol::EvictionReason::NoSession,
+ );
+ if let Err(error) = shard
+ .bus
+ .send_to_client(transport_client_id,
eviction.into_generic().into_frozen())
+ .await
+ {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "failed to send unauthenticated eviction"
+ );
+ }
+}
+
+/// Serve `poll_messages`: resolve the partition namespace, run the read on
+/// the owning shard ([`shard::IggyShard::partition_read`]), and re-encode
+/// the stored batches into the legacy wire `PolledMessages` body.
+///
+/// Failures reply with an empty body so the SDK fails fast on decode
+/// instead of hanging until its read timeout.
+#[allow(clippy::future_not_send)]
+async fn handle_poll_messages(
+ shard: &Rc<ServerNgShard>,
+ transport_client_id: u128,
+ request: &Message<RequestHeader>,
+) {
+ let body = match decode_poll_request(shard, request) {
+ Ok((namespace, partition_id, consumer, args)) => {
+ match shard
+ .partition_read(namespace, PartitionRead::Poll { consumer,
args })
+ .await
+ {
+ Some(PartitionReadReply::Poll {
+ fragments,
+ current_offset,
+ }) => build_polled_messages_body(partition_id, current_offset,
fragments)
+ .unwrap_or_else(|error| {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "failed to re-encode polled batches; replying
empty poll"
+ );
+ empty_polled_messages_body(partition_id)
+ }),
+ other => {
+ warn!(
+ transport_client_id,
+ namespace = namespace.inner(),
+ reply_was_none = other.is_none(),
+ "partition read failed; replying empty poll"
+ );
+ empty_polled_messages_body(partition_id)
+ }
+ }
+ }
+ Err(error) => {
+ // A zero-byte body would panic the SDK's `PolledMessages`
+ // decoder; reply the 16-byte empty-poll shape instead.
+ warn!(
+ transport_client_id,
+ error = %error,
+ "poll_messages request rejected; replying empty poll"
+ );
+ empty_polled_messages_body(0)
+ }
+ };
+ send_non_replicated_bytes(shard, request, transport_client_id, body,
"poll_messages").await;
+}
+
+/// Serve `get_consumer_offset`. An empty body decodes as `None` on the SDK
+/// side (no offset stored / partition unknown).
+#[allow(clippy::future_not_send)]
+async fn handle_get_consumer_offset(
+ shard: &Rc<ServerNgShard>,
+ transport_client_id: u128,
+ request: &Message<RequestHeader>,
+) {
+ let body = match decode_consumer_offset_request(shard, request) {
+ Ok((namespace, partition_id, consumer)) => {
+ match shard
+ .partition_read(namespace, PartitionRead::ConsumerOffset {
consumer })
+ .await
+ {
+ Some(PartitionReadReply::ConsumerOffset {
+ stored: Some(stored_offset),
+ current_offset,
+ }) => build_consumer_offset_body(partition_id, current_offset,
stored_offset),
+ _ => Bytes::new(),
+ }
+ }
+ Err(error) => {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "get_consumer_offset request rejected; replying empty"
+ );
+ Bytes::new()
+ }
+ };
+ send_non_replicated_bytes(
+ shard,
+ request,
+ transport_client_id,
+ body,
+ "get_consumer_offset",
+ )
+ .await;
+}
+
+/// Wait (bounded) until `namespace` is routable: this shard's routing row
+/// exists and the owning shard answers a probe read (partition
+/// materialised). Fast path: row already present -> no probe, no wait.
+///
+/// Covers the post-`CreateTopic` convergence window where the metadata
+/// commit has returned to the client but the per-shard reconcilers have
+/// not yet seeded routing rows / materialised partitions.
+#[allow(clippy::future_not_send)]
+/// Ack a partition op that cannot be routed (unresolved or never-
+/// materialised namespace) with an empty Reply. The SDK connection
+/// processes replies in lockstep, so a silent drop wedges every
+/// subsequent request on that connection.
+async fn send_empty_partition_reply(
+ shard: &Rc<ServerNgShard>,
+ transport_client_id: u128,
+ request_header: &RequestHeader,
+) {
+ let commit = current_metadata_commit(shard);
+ let reply = build_empty_reply(request_header, transport_client_id, 0,
commit);
+ if let Err(error) = shard
+ .bus
+ .send_to_client(transport_client_id,
reply.into_generic().into_frozen())
+ .await
+ {
+ warn!(
+ transport_client_id,
+ error = %error,
+ operation = ?request_header.operation,
+ "failed to surface empty partition reply"
+ );
+ }
+}
+
+async fn wait_for_partition_routable(shard: &Rc<ServerNgShard>, namespace:
IggyNamespace) -> bool {
+ const ATTEMPT_DELAY: std::time::Duration =
std::time::Duration::from_millis(50);
+ const BUDGET: std::time::Duration = std::time::Duration::from_secs(3);
+
+ if shard.shards_table().shard_for(namespace).is_some() {
+ return true;
+ }
+ let deadline = std::time::Instant::now() + BUDGET;
+ while shard.shards_table().shard_for(namespace).is_none() {
+ if std::time::Instant::now() >= deadline {
+ return false;
+ }
+ compio::time::sleep(ATTEMPT_DELAY).await;
+ }
+ // The local row is seeded by THIS shard's reconciler; the owner
+ // materialises the partition on its own pass. Probe with a cheap read
+ // until the owner answers, so the write below cannot be dropped by the
Review Comment:
the comment says the write `cannot be dropped by the owner's guard`, but the
partition can de-materialise between this probe returning `Some` and the
dispatch below - it's just benign because the downstream park/tombstone path
re-checks and the client retries. worth softening the comment so it doesn't
read as a hard guarantee.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -965,10 +1246,37 @@ where
return;
}
} else {
- debug_assert_eq!(
- header.op, current_op,
- "primary: sequencer pre-advance broken"
+ // Primary: `push_prepare_entry` pre-advanced the sequencer, so a
+ // locally-originated prepare always satisfies
+ // `header.op == current_op`. The two violation directions carry
+ // very different risk:
+ // - below the sequencer: a duplicate delivery (parked-frame
+ // redispatch, retransmit echo) of an op this primary already
+ // sequenced. Apply is keyed by `header.op` and the primary
+ // never advances its sequencer post-apply, so proceeding is
+ // idempotent-safe; log loudly for diagnosis.
+ // - above the sequencer: journaling an op the sequencer has not
+ // assigned yet means the next local assignment would collide
+ // with it. Not recoverable in place; crash in release too
+ // rather than corrupt op assignment silently.
+ assert!(
Review Comment:
this is a release `assert!` in a library on a should-never-happen path.
`header.op > current_op` looks unreachable today (the `replicate_preflight`
view fences run first, one primary per view, and the chain ring stops before
the primary), so prefer `debug_assert!` plus a graceful log-and-return here -
apply is keyed by `header.op` and idempotent. separately: the `header.op <
current_op` idempotent-apply branch added just below is only reachable when a
prior apply left the sequencer ahead of the journal, which is exactly the
`write_append` cursor bug - so this branch masks that symptom rather than
fixing it. the metadata plane keeps the tighter `debug_assert_eq` for the same
invariant; better to fix the journal cursor and keep both planes strict than to
relax the partition side.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -674,6 +687,269 @@ where
IggyNamespace::from_raw(self.consensus.namespace())
}
+ /// Resolve a poll query against the in-memory journal, falling back to
+ /// the on-disk segments for ranges the journal no longer holds (the
+ /// persist threshold drains committed batches to segment files).
+ ///
+ /// A query is served from exactly one tier per call: a poll that starts
+ /// below the journal's oldest resident offset reads from disk only, and
+ /// the client's next poll (advancing past what was returned) eventually
+ /// crosses back into the resident range. Timestamp queries try disk
+ /// first whenever segments hold persisted bytes -- older matches always
+ /// live there -- and fall back to the journal when the disk has none.
+ async fn lookup_messages(&self, query: MessageLookup) ->
Option<PollQueryResult<4096>> {
+ let serve_journal_first = match query {
+ MessageLookup::Offset { offset, .. } => self
+ .log
+ .journal()
+ .inner
+ .oldest_resident_offset()
+ .is_some_and(|oldest| offset >= oldest),
+ MessageLookup::Timestamp { .. } =>
!self.has_persisted_segment_bytes(),
+ };
+
+ if serve_journal_first {
+ return self.log.journal().inner.get(&query).await;
+ }
+ match self.poll_from_disk(query).await {
+ Some((mut fragments, last_matching_offset, matched)) => {
+ // A poll can straddle the tiers: older messages already
+ // drained to segments, the tail still journal-resident.
+ // Continue past the last disk match by offset (timestamp
+ // matches are contiguous from the first hit, so an offset
+ // continuation is equivalent).
+ let remaining = query.count().saturating_sub(matched);
+ if remaining > 0
+ && let Some(last_offset) = last_matching_offset
+ {
+ let continuation = MessageLookup::Offset {
+ offset: last_offset + 1,
+ count: remaining,
+ };
+ if let Some((journal_fragments, journal_last)) =
+ self.log.journal().inner.get(&continuation).await
+ {
+ fragments.extend(journal_fragments);
+ return Some((fragments,
journal_last.or(last_matching_offset)));
+ }
+ }
+ Some((fragments, last_matching_offset))
+ }
+ // Nothing matched on disk (e.g. a timestamp newer than every
+ // persisted batch): the match, if any, is journal-resident.
+ None => self.log.journal().inner.get(&query).await,
+ }
+ }
+
+ fn partition_dir(&self) -> Option<String> {
+ if self.partition_dir.is_some() {
+ return self.partition_dir.clone();
+ }
+ // Writer-derived fallback for partitions built without
+ // `set_partition_dir`. Unreliable mid-rotation: sealed segments
+ // drop their writer, so prefer the stored path above.
+ self.log
+ .messages_writers()
+ .iter()
+ .rev()
+ .flatten()
+ .next()
+ .and_then(|writer| {
+ std::path::Path::new(&writer.path())
+ .parent()
+ .map(|dir| dir.to_string_lossy().into_owned())
+ })
+ }
+
+ fn has_persisted_segment_bytes(&self) -> bool {
+ self.log
+ .segments()
+ .iter()
+ .any(|segment| segment.size.as_bytes_u64() > 0)
+ }
+
+ /// Serve a poll from the on-disk segment files.
+ ///
+ /// Picks the starting segment + byte position via the sparse index
+ /// (one entry per persist flush; a miss falls back to the segment
+ /// start), then walks stamped `[256B SendMessages2Header][blob]`
+ /// batches in chunked reads, slicing fragments with the same selector
+ /// the journal path uses. Batches split across a chunk boundary are
+ /// re-read from their start in the next chunk.
+ #[allow(clippy::cast_possible_truncation)]
+ async fn poll_from_disk(
+ &self,
+ query: MessageLookup,
+ ) -> Option<(PollFragments<4096>, Option<u64>, u32)> {
+ const DISK_POLL_CHUNK: u64 = 1 << 20;
+
+ let count = query.count();
+ if count == 0 || !self.log.has_segments() {
+ return None;
+ }
+
+ let (start_segment, mut position) = self.disk_poll_start(&query);
+
+ let mut fragments = PollFragments::new();
+ let mut last_matching_offset = None;
+ let mut matched: u32 = 0;
+
+ for segment_index in start_segment..self.log.segments().len() {
+ if matched >= count {
+ break;
+ }
+ let persisted =
self.log.segments()[segment_index].size.as_bytes_u64();
+ if persisted == 0 || position >= persisted {
+ position = 0;
+ continue;
+ }
+ // Sealed segments drop their writer at rotation, so resolve the
+ // file from the partition directory (taken from any live writer)
+ // plus the segment's start offset, mirroring the writer naming.
+ let Some(partition_dir) = self.partition_dir() else {
+ // Simulated in-memory persistence: no files to read. A live
+ // partition hitting this means no writer was resolvable
+ // (e.g. mid-rotation), which silently hides the disk tier.
+ warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = self.namespace().inner(),
+ segment_count = self.log.segments().len(),
+ "disk poll: no live writer to resolve partition dir; disk
tier unreadable"
+ );
+ return None;
+ };
+ let start_offset = self.log.segments()[segment_index].start_offset;
+ let path = format!("{partition_dir}/{start_offset:0>20}.log");
+ let Some(file) = self.open_segment_with_retry(&path).await else {
+ position = 0;
+ continue;
+ };
+
+ let mut chunk_len = DISK_POLL_CHUNK;
+ while matched < count && position < persisted {
+ let len = (persisted - position).min(chunk_len) as usize;
+ let Some(chunk) = self.read_chunk_with_retry(&file, position,
len).await else {
+ break;
+ };
+ let consumed = walk_disk_chunk(
+ &chunk,
+ query,
+ count,
+ &mut matched,
+ &mut fragments,
+ &mut last_matching_offset,
+ );
+ if consumed == 0 {
+ if (len as u64) >= persisted - position {
+ // The whole remainder fit and still no complete
+ // batch decoded: corrupt tail; stop.
+ break;
+ }
+ // A single batch larger than the chunk: grow and
+ // re-read instead of spinning.
+ chunk_len = chunk_len.saturating_mul(4);
Review Comment:
when a batch is bigger than the chunk this regrows by 4x and re-reads from
`position`, discarding the bytes just read. the batch header already carries
`total_size`, so size the next read to the actual batch length once the header
decodes, instead of a blind 4x grow-and-reread.
##########
core/server-ng/src/responses.rs:
##########
@@ -744,3 +746,121 @@ pub(crate) fn current_metadata_commit(shard:
&Rc<ServerNgShard>) -> u64 {
.as_ref()
.map_or(0, VsrConsensus::commit_max)
}
+
+/// 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`.
+const STORED_MESSAGE_HEADER_SIZE: usize = 48;
+
+/// Build the `PolledMessages` reply body from the owning shard's poll
+/// fragments.
+///
+/// Fragments carry the stored `SendMessages2` batches: a 256-byte command
+/// header followed by `IggyMessage2`-format messages
+/// (`[48B header][payload][user_headers]`, offsets/timestamps delta-encoded
+/// against the batch). The SDK decodes the legacy wire format
+/// (`[64B header][payload][user_headers]`, absolute offsets); the message
+/// sections share the legacy order, so only the header is re-encoded here
+/// and the section bytes copy through contiguously.
+///
+/// Body layout: `[partition_id:4][current_offset:8][count:4][messages...]`.
+pub(crate) fn build_polled_messages_body(
+ partition_id: u32,
+ current_offset: u64,
+ fragments: PollFragments,
+) -> Result<Bytes, IggyError> {
+ // Batches may arrive split across fragments (rewritten command header +
+ // sliced blob); concatenate into one stream before walking batches.
+ let mut stream: Vec<u8> = Vec::new();
Review Comment:
this copies the whole poll payload twice on every poll: once concatenating
all fragments into `stream`, then again re-encoding message-by-message into
`messages` (plus the final `body` alloc). the fragments are refcounted `Frozen`
slices, so the concat throws away the zero-copy. since only the 16-byte header
differs from the stored format, you can walk the fragments in place and emit
`[rewritten header][payload+user_headers slice]` as chained `Bytes` without the
two intermediate buffers.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -1870,3 +2200,59 @@ where
send_prepare_ok_common(self.consensus(), header, Some(true)).await;
}
}
+
+/// Walk stamped `[256B SendMessages2Header][blob]` batches in one disk
+/// chunk, pushing matching fragments. Returns bytes consumed: the start
+/// of the first batch that did not fully fit in the chunk (the caller
+/// re-reads from there), or the chunk end when everything decoded.
+fn walk_disk_chunk(
+ chunk: &Frozen<4096>,
+ query: MessageLookup,
+ count: u32,
+ matched: &mut u32,
+ fragments: &mut PollFragments<4096>,
+ last_matching_offset: &mut Option<u64>,
+) -> usize {
+ let bytes: &[u8] = chunk;
+ let mut cursor = 0usize;
+
+ while *matched < count && cursor + COMMAND_HEADER_SIZE <= bytes.len() {
+ let Ok(batch) = decode_batch_slice(&bytes[cursor..]) else {
Review Comment:
disk poll path: `decode_batch_slice` validates the full batch checksum (an
xxhash pass over the whole blob) here, then `build_polled_messages_body` walks
the same bytes again header-only to re-encode. that's a double parse plus a
hash the response builder doesn't need. could have the disk walk emit
already-rewritten wire messages, or a header-only decode variant that only
checksums when integrity is actually wanted. disk-tier only (polls below the
resident range), so not steady-state, but real on backlog replay.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -674,6 +687,269 @@ where
IggyNamespace::from_raw(self.consensus.namespace())
}
+ /// Resolve a poll query against the in-memory journal, falling back to
+ /// the on-disk segments for ranges the journal no longer holds (the
+ /// persist threshold drains committed batches to segment files).
+ ///
+ /// A query is served from exactly one tier per call: a poll that starts
+ /// below the journal's oldest resident offset reads from disk only, and
+ /// the client's next poll (advancing past what was returned) eventually
+ /// crosses back into the resident range. Timestamp queries try disk
+ /// first whenever segments hold persisted bytes -- older matches always
+ /// live there -- and fall back to the journal when the disk has none.
+ async fn lookup_messages(&self, query: MessageLookup) ->
Option<PollQueryResult<4096>> {
+ let serve_journal_first = match query {
+ MessageLookup::Offset { offset, .. } => self
+ .log
+ .journal()
+ .inner
+ .oldest_resident_offset()
+ .is_some_and(|oldest| offset >= oldest),
+ MessageLookup::Timestamp { .. } =>
!self.has_persisted_segment_bytes(),
+ };
+
+ if serve_journal_first {
+ return self.log.journal().inner.get(&query).await;
+ }
+ match self.poll_from_disk(query).await {
+ Some((mut fragments, last_matching_offset, matched)) => {
+ // A poll can straddle the tiers: older messages already
+ // drained to segments, the tail still journal-resident.
+ // Continue past the last disk match by offset (timestamp
+ // matches are contiguous from the first hit, so an offset
+ // continuation is equivalent).
+ let remaining = query.count().saturating_sub(matched);
+ if remaining > 0
+ && let Some(last_offset) = last_matching_offset
+ {
+ let continuation = MessageLookup::Offset {
+ offset: last_offset + 1,
+ count: remaining,
+ };
+ if let Some((journal_fragments, journal_last)) =
+ self.log.journal().inner.get(&continuation).await
+ {
+ fragments.extend(journal_fragments);
+ return Some((fragments,
journal_last.or(last_matching_offset)));
+ }
+ }
+ Some((fragments, last_matching_offset))
+ }
+ // Nothing matched on disk (e.g. a timestamp newer than every
+ // persisted batch): the match, if any, is journal-resident.
+ None => self.log.journal().inner.get(&query).await,
+ }
+ }
+
+ fn partition_dir(&self) -> Option<String> {
+ if self.partition_dir.is_some() {
+ return self.partition_dir.clone();
+ }
+ // Writer-derived fallback for partitions built without
+ // `set_partition_dir`. Unreliable mid-rotation: sealed segments
+ // drop their writer, so prefer the stored path above.
+ self.log
+ .messages_writers()
+ .iter()
+ .rev()
+ .flatten()
+ .next()
+ .and_then(|writer| {
+ std::path::Path::new(&writer.path())
+ .parent()
+ .map(|dir| dir.to_string_lossy().into_owned())
+ })
+ }
+
+ fn has_persisted_segment_bytes(&self) -> bool {
+ self.log
+ .segments()
+ .iter()
+ .any(|segment| segment.size.as_bytes_u64() > 0)
+ }
+
+ /// Serve a poll from the on-disk segment files.
+ ///
+ /// Picks the starting segment + byte position via the sparse index
+ /// (one entry per persist flush; a miss falls back to the segment
+ /// start), then walks stamped `[256B SendMessages2Header][blob]`
+ /// batches in chunked reads, slicing fragments with the same selector
+ /// the journal path uses. Batches split across a chunk boundary are
+ /// re-read from their start in the next chunk.
+ #[allow(clippy::cast_possible_truncation)]
+ async fn poll_from_disk(
+ &self,
+ query: MessageLookup,
+ ) -> Option<(PollFragments<4096>, Option<u64>, u32)> {
+ const DISK_POLL_CHUNK: u64 = 1 << 20;
+
+ let count = query.count();
+ if count == 0 || !self.log.has_segments() {
+ return None;
+ }
+
+ let (start_segment, mut position) = self.disk_poll_start(&query);
+
+ let mut fragments = PollFragments::new();
+ let mut last_matching_offset = None;
+ let mut matched: u32 = 0;
+
+ for segment_index in start_segment..self.log.segments().len() {
+ if matched >= count {
+ break;
+ }
+ let persisted =
self.log.segments()[segment_index].size.as_bytes_u64();
+ if persisted == 0 || position >= persisted {
+ position = 0;
+ continue;
+ }
+ // Sealed segments drop their writer at rotation, so resolve the
+ // file from the partition directory (taken from any live writer)
+ // plus the segment's start offset, mirroring the writer naming.
+ let Some(partition_dir) = self.partition_dir() else {
+ // Simulated in-memory persistence: no files to read. A live
+ // partition hitting this means no writer was resolvable
+ // (e.g. mid-rotation), which silently hides the disk tier.
+ warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ namespace_raw = self.namespace().inner(),
+ segment_count = self.log.segments().len(),
+ "disk poll: no live writer to resolve partition dir; disk
tier unreadable"
+ );
+ return None;
+ };
+ let start_offset = self.log.segments()[segment_index].start_offset;
+ let path = format!("{partition_dir}/{start_offset:0>20}.log");
+ let Some(file) = self.open_segment_with_retry(&path).await else {
+ position = 0;
+ continue;
+ };
+
+ let mut chunk_len = DISK_POLL_CHUNK;
+ while matched < count && position < persisted {
+ let len = (persisted - position).min(chunk_len) as usize;
+ let Some(chunk) = self.read_chunk_with_retry(&file, position,
len).await else {
+ break;
+ };
+ let consumed = walk_disk_chunk(
+ &chunk,
+ query,
+ count,
+ &mut matched,
+ &mut fragments,
+ &mut last_matching_offset,
+ );
+ if consumed == 0 {
+ if (len as u64) >= persisted - position {
+ // The whole remainder fit and still no complete
+ // batch decoded: corrupt tail; stop.
+ break;
+ }
+ // A single batch larger than the chunk: grow and
+ // re-read instead of spinning.
+ chunk_len = chunk_len.saturating_mul(4);
+ continue;
+ }
+ chunk_len = DISK_POLL_CHUNK;
+ position += consumed as u64;
+ }
+ position = 0;
+ }
+
+ if fragments.is_empty() {
+ None
+ } else {
+ Some((fragments, last_matching_offset, matched))
+ }
+ }
+
+ /// 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.
+ async fn open_segment_with_retry(&self, path: &str) ->
Option<compio::fs::File> {
+ for attempt in 0..3u8 {
+ match compio::fs::File::open(path).await {
Review Comment:
disk poll opens a fresh file per segment per poll (no fd cache) and
re-resolves plus clones `partition_dir()` once per segment. a from-offset-0
replay over many small segments is then one open and one path alloc per
segment, every poll. sealed segments are immutable so their fds could be
cached, and `partition_dir()` can be hoisted out of the segment loop.
##########
core/journal/src/file_storage.rs:
##########
@@ -89,19 +89,33 @@ impl FileStorage {
Ok(buf)
}
- /// Append write, returns bytes written.
+ /// Append write at the next free offset; returns the offset written to.
+ ///
+ /// Reserves the region by advancing `write_offset` **synchronously,
+ /// before** the write `.await`. On the single-threaded compio runtime
+ /// two `write_append` calls can interleave at the await; reserving first
+ /// hands each a distinct, non-overlapping offset. The previous code read
+ /// the offset and advanced the cursor *after* the await, so two in-flight
+ /// appends both saw the same offset and wrote over each other (and the
+ /// journal recorded the same index position for both) -- the corruption
+ /// seen under interleaved `on_replicate` calls (a queued op drained while
+ /// the next op is freshly submitted).
+ ///
+ /// The reservation is not rolled back on write error: the space stays
+ /// reserved, so the failing op leaves an uncommitted gap that recovery
+ /// truncates rather than a slot a later append could reuse mid-flight.
///
/// # Errors
/// Returns an I/O error if the write fails.
- #[allow(clippy::cast_possible_truncation)]
- pub async fn write_append<B: IoBuf>(&self, buf: B) -> io::Result<usize> {
- let len = buf.buf_len();
+ pub async fn write_append<B: IoBuf>(&self, buf: B) -> io::Result<u64> {
+ let len = buf.buf_len() as u64;
+ let offset = self.write_offset.get();
+ self.write_offset.set(offset + len);
Review Comment:
`write_append` advances `write_offset` here before the write `.await`, but
never rolls it back if `write_all_at` fails. on a write error the `?` returns
yet the cursor stays advanced. `on_replicate` only logs and returns on append
failure (it doesn't crash or poison), so the inflated cursor survives on the
long-lived `FileStorage`. the next append - e.g. a vsr retransmit of the op -
then writes past a zero hole, fsyncs and acks. on reopen the recovery scan hits
the zero bytes at the hole, treats it as a corrupt/non-prepare entry and
truncates from there, silently dropping the later op that was already committed
and fsync'd. since this is the metadata consensus WAL, that's loss of committed
metadata (streams/users/PATs) plus cluster divergence, from a single write
fault. the `recovery truncates the gap` note above assumes a reclaimable tail,
but the hole sits under a committed op. fix: roll back `write_offset` to
`offset` on the error path, or write a skippable filler record so the
scan steps over instead of truncating.
##########
core/server-ng/src/dispatch.rs:
##########
@@ -606,6 +763,293 @@ async fn send_non_replicated_bytes(
}
}
+/// Reject a pre-auth request with a typed `Eviction(NoSession)` frame.
+///
+/// The SDK's reply decoder maps eviction reasons to typed errors
+/// (`NoSession` -> `Unauthenticated`), so clients fail fast with the same
+/// error the legacy server returns instead of a body-decode failure. The
+/// eviction context is best-effort off the metadata consensus (peer shards
+/// have none; zeroes are cosmetic -- the SDK only reads the reason).
+#[allow(clippy::future_not_send)]
+async fn send_unauthenticated_eviction(shard: &Rc<ServerNgShard>,
transport_client_id: u128) {
+ let ctx = shard.plane.metadata().consensus.as_ref().map_or(
+ consensus::EvictionContext {
+ cluster: 0,
+ view: 0,
+ replica: 0,
+ },
+ consensus::EvictionContext::from_consensus,
+ );
+ let eviction = consensus::build_eviction_message(
+ ctx,
+ transport_client_id,
+ iggy_binary_protocol::EvictionReason::NoSession,
+ );
+ if let Err(error) = shard
+ .bus
+ .send_to_client(transport_client_id,
eviction.into_generic().into_frozen())
+ .await
+ {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "failed to send unauthenticated eviction"
+ );
+ }
+}
+
+/// Serve `poll_messages`: resolve the partition namespace, run the read on
+/// the owning shard ([`shard::IggyShard::partition_read`]), and re-encode
+/// the stored batches into the legacy wire `PolledMessages` body.
+///
+/// Failures reply with an empty body so the SDK fails fast on decode
+/// instead of hanging until its read timeout.
+#[allow(clippy::future_not_send)]
+async fn handle_poll_messages(
+ shard: &Rc<ServerNgShard>,
+ transport_client_id: u128,
+ request: &Message<RequestHeader>,
+) {
+ let body = match decode_poll_request(shard, request) {
+ Ok((namespace, partition_id, consumer, args)) => {
+ match shard
+ .partition_read(namespace, PartitionRead::Poll { consumer,
args })
+ .await
+ {
+ Some(PartitionReadReply::Poll {
+ fragments,
+ current_offset,
+ }) => build_polled_messages_body(partition_id, current_offset,
fragments)
+ .unwrap_or_else(|error| {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "failed to re-encode polled batches; replying
empty poll"
+ );
+ empty_polled_messages_body(partition_id)
+ }),
+ other => {
+ warn!(
+ transport_client_id,
+ namespace = namespace.inner(),
+ reply_was_none = other.is_none(),
+ "partition read failed; replying empty poll"
+ );
+ empty_polled_messages_body(partition_id)
+ }
+ }
+ }
+ Err(error) => {
+ // A zero-byte body would panic the SDK's `PolledMessages`
+ // decoder; reply the 16-byte empty-poll shape instead.
+ warn!(
+ transport_client_id,
+ error = %error,
+ "poll_messages request rejected; replying empty poll"
+ );
+ empty_polled_messages_body(0)
+ }
+ };
+ send_non_replicated_bytes(shard, request, transport_client_id, body,
"poll_messages").await;
+}
+
+/// Serve `get_consumer_offset`. An empty body decodes as `None` on the SDK
+/// side (no offset stored / partition unknown).
+#[allow(clippy::future_not_send)]
+async fn handle_get_consumer_offset(
+ shard: &Rc<ServerNgShard>,
+ transport_client_id: u128,
+ request: &Message<RequestHeader>,
+) {
+ let body = match decode_consumer_offset_request(shard, request) {
+ Ok((namespace, partition_id, consumer)) => {
+ match shard
+ .partition_read(namespace, PartitionRead::ConsumerOffset {
consumer })
+ .await
+ {
+ Some(PartitionReadReply::ConsumerOffset {
+ stored: Some(stored_offset),
+ current_offset,
+ }) => build_consumer_offset_body(partition_id, current_offset,
stored_offset),
+ _ => Bytes::new(),
+ }
+ }
+ Err(error) => {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "get_consumer_offset request rejected; replying empty"
+ );
+ Bytes::new()
+ }
+ };
+ send_non_replicated_bytes(
+ shard,
+ request,
+ transport_client_id,
+ body,
+ "get_consumer_offset",
+ )
+ .await;
+}
+
+/// Wait (bounded) until `namespace` is routable: this shard's routing row
+/// exists and the owning shard answers a probe read (partition
+/// materialised). Fast path: row already present -> no probe, no wait.
+///
+/// Covers the post-`CreateTopic` convergence window where the metadata
+/// commit has returned to the client but the per-shard reconcilers have
+/// not yet seeded routing rows / materialised partitions.
+#[allow(clippy::future_not_send)]
+/// Ack a partition op that cannot be routed (unresolved or never-
+/// materialised namespace) with an empty Reply. The SDK connection
+/// processes replies in lockstep, so a silent drop wedges every
+/// subsequent request on that connection.
+async fn send_empty_partition_reply(
+ shard: &Rc<ServerNgShard>,
+ transport_client_id: u128,
+ request_header: &RequestHeader,
+) {
+ let commit = current_metadata_commit(shard);
+ let reply = build_empty_reply(request_header, transport_client_id, 0,
commit);
+ if let Err(error) = shard
+ .bus
+ .send_to_client(transport_client_id,
reply.into_generic().into_frozen())
+ .await
+ {
+ warn!(
+ transport_client_id,
+ error = %error,
+ operation = ?request_header.operation,
+ "failed to surface empty partition reply"
+ );
+ }
+}
+
+async fn wait_for_partition_routable(shard: &Rc<ServerNgShard>, namespace:
IggyNamespace) -> bool {
+ const ATTEMPT_DELAY: std::time::Duration =
std::time::Duration::from_millis(50);
+ const BUDGET: std::time::Duration = std::time::Duration::from_secs(3);
+
+ if shard.shards_table().shard_for(namespace).is_some() {
+ return true;
+ }
+ let deadline = std::time::Instant::now() + BUDGET;
+ while shard.shards_table().shard_for(namespace).is_none() {
+ if std::time::Instant::now() >= deadline {
+ return false;
+ }
+ compio::time::sleep(ATTEMPT_DELAY).await;
Review Comment:
the routable probe fires a full cross-shard read roundtrip every 50ms during
the post-create-topic convergence window. only hits that window so it's minor,
but a single-flight probe per namespace or a routability signal from the
reconciler would avoid the spam when many clients write to a just-created topic
at once.
##########
core/partitions/src/journal.rs:
##########
@@ -474,7 +481,7 @@ impl QueryableJournal<PartitionJournalMemStorage> for
PartitionJournal<Partition
}
}
-fn select_batch_slice(
+pub fn select_batch_slice(
Review Comment:
`select_batch_slice`, `SelectedBatchSlice` (line 53) and
`oldest_resident_offset` (264) - plus `segment_indexes` in `log.rs:213` - are
all `pub` but every caller is inside the `partitions` crate, so `pub(crate)`
fits. `decode_batch_slice` is the one that genuinely needs `pub` (cross-crate).
`count` (line 45) is borderline since it's on a pub type.
##########
core/server-ng/src/bootstrap.rs:
##########
@@ -864,6 +865,27 @@ async fn shard_main(
});
bus.track_background(reconciler_handle);
+ // Consensus timer driver: heartbeats, prepare retransmit, and
+ // view-change timeouts only advance when `VsrConsensus::tick` runs
+ // ("call this periodically, e.g. every 10ms"). The simulator steps it
+ // explicitly; production drives it here. Without this, a prepare lost
+ // to a transient replica-link blip is never retransmitted and its
+ // client request hangs until the SDK read timeout.
+ let (consensus_tick_stop_tx, consensus_tick_stop_rx) = channel::<()>(1);
+ let tick_shard = Rc::clone(&shard);
+ let consensus_tick_handle = compio::runtime::spawn(async move {
+ const CONSENSUS_TICK_INTERVAL: std::time::Duration =
std::time::Duration::from_millis(10);
+ loop {
+ if consensus_tick_stop_rx.try_recv().is_ok() {
+ break;
+ }
+ tick_shard.tick_metadata().await;
Review Comment:
this tick wakes every 10ms per shard forever and `tick_partitions` builds a
fresh `namespaces()` Vec every tick even when there's no armed timeout to
service - idle cost scales with partition count. fine for now, but a follow-up
could rearm event-driven / skip when nothing is pending (matches the
tigerbeetle timer-lifecycle TODO in consensus).
##########
core/sdk/src/tcp/tcp_client.rs:
##########
@@ -719,7 +755,18 @@ impl TcpClient {
let body_size = response_size -
iggy_binary_protocol::HEADER_SIZE;
let body = if body_size > 0 {
let mut body = BytesMut::with_capacity(body_size);
- stream.read_buf(&mut body,
body_size).await.map_err(|error| {
+ let body_read = tokio::time::timeout(
+ RESPONSE_READ_TIMEOUT,
Review Comment:
the header read and body read each get their own full
`RESPONSE_READ_TIMEOUT`, so a reply that delivers a header then stalls can wait
up to 2x the timeout total instead of 30s. use a single shared deadline
(`Instant::now() + RESPONSE_READ_TIMEOUT`) across both reads.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
