You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/06/17 10:14:01 UTC
(iggy) branch fix_integration_tests_partitions_permissions updated: code review nits
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch fix_integration_tests_partitions_permissions
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/fix_integration_tests_partitions_permissions by this push:
new c003784ec code review nits
c003784ec is described below
commit c003784ec9110bdae1854b24eb6c83cb02fd52d9
Author: Grzegorz Koszyk <g....@gmail.com>
AuthorDate: Wed Jun 17 12:13:55 2026 +0200
code review nits
---
core/journal/src/file_storage.rs | 19 +++++++++++++----
core/partitions/src/iggy_partition.rs | 30 +++++++++++++++++++-------
core/sdk/src/tcp/tcp_client.rs | 18 ++++++++++------
core/sdk/src/websocket/websocket_client.rs | 34 ++++++++++++++++++++++++++++--
core/server-ng/src/dispatch.rs | 23 +++++++++++---------
core/server-ng/src/responses.rs | 25 ++++++----------------
core/server_common/src/send_messages2.rs | 31 ++++-----------------------
core/shard/src/lib.rs | 25 ++++++++++++++++++++++
8 files changed, 129 insertions(+), 76 deletions(-)
diff --git a/core/journal/src/file_storage.rs b/core/journal/src/file_storage.rs
index 81f1a983f..b6af78277 100644
--- a/core/journal/src/file_storage.rs
+++ b/core/journal/src/file_storage.rs
@@ -101,9 +101,15 @@ impl FileStorage {
/// 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.
+ /// On write error the reservation is rolled back **only when no later
+ /// append has reserved space after us** (`write_offset == offset + len`).
+ /// Leaving the gap is unsafe: a subsequent append (e.g. a VSR retransmit
+ /// of this op) would write past the zero hole, fsync and ack, and on
+ /// reopen the recovery scan would hit the hole and truncate from there --
+ /// silently dropping the committed op that now sits above the gap. Rolling
+ /// back lets the next append reuse the slot instead. In the rare
+ /// interleave where another append already reserved after us, rolling back
+ /// would clobber its slot, so the gap is left for recovery in that case.
///
/// # Errors
/// Returns an I/O error if the write fails.
@@ -114,7 +120,12 @@ impl FileStorage {
// SAFETY: single-threaded compio runtime, no concurrent access to the file.
let file = unsafe { &mut *self.file.get() };
let (result, _buf) = file.write_all_at(buf, offset).await.into();
- result?;
+ if let Err(error) = result {
+ if self.write_offset.get() == offset + len {
+ self.write_offset.set(offset);
+ }
+ return Err(error);
+ }
Ok(offset)
}
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index 239cc4ea0..029fd35fd 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -1257,14 +1257,28 @@ where
// 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!(
- header.op <= current_op,
- "primary: prepare op {} ahead of sequencer {}; next op assignment would collide",
- header.op,
- current_op
- );
+ // with it. Unreachable today (view fences run first, one
+ // primary per view, the chain stops before the primary), so
+ // trip the invariant in debug; in release log loudly and drop
+ // rather than crash a library or corrupt op assignment.
+ if header.op > current_op {
+ debug_assert!(
+ header.op <= current_op,
+ "primary: prepare op {} ahead of sequencer {}; next op assignment would collide",
+ header.op,
+ current_op
+ );
+ emit_partition_diag(
+ tracing::Level::ERROR,
+ &PartitionDiagEvent::new(
+ self.diag_ctx(),
+ "primary prepare ahead of sequencer; dropping to avoid op-assignment collision",
+ )
+ .with_operation(header.operation)
+ .with_op(header.op),
+ );
+ return;
+ }
if header.op < current_op {
emit_partition_diag(
tracing::Level::WARN,
diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs
index 2a66d3c84..f6e6f4980 100644
--- a/core/sdk/src/tcp/tcp_client.rs
+++ b/core/sdk/src/tcp/tcp_client.rs
@@ -730,11 +730,15 @@ impl TcpClient {
// hold the stream lock forever and wedge every later
// request on this client. On expiry drop the stream --
// a late reply would desync framing for the next request.
- let header_read = tokio::time::timeout(
- RESPONSE_READ_TIMEOUT,
- stream.read(&mut response_header),
- )
- .await;
+ //
+ // One deadline spans BOTH the header and body reads: a
+ // reply that delivers a header then stalls must not get a
+ // fresh full timeout for the body (which would allow up to
+ // 2x `RESPONSE_READ_TIMEOUT` total).
+ let response_deadline = tokio::time::Instant::now() + RESPONSE_READ_TIMEOUT;
+ let header_read =
+ tokio::time::timeout_at(response_deadline, stream.read(&mut response_header))
+ .await;
let Ok(header_read) = header_read else {
error!(
"Timed out after {RESPONSE_READ_TIMEOUT:?} waiting for VSR response header for TCP request with code: {code}",
@@ -755,8 +759,8 @@ 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);
- let body_read = tokio::time::timeout(
- RESPONSE_READ_TIMEOUT,
+ let body_read = tokio::time::timeout_at(
+ response_deadline,
stream.read_buf(&mut body, body_size),
)
.await;
diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs
index 5e5e35e2a..e1a983839 100644
--- a/core/sdk/src/websocket/websocket_client.rs
+++ b/core/sdk/src/websocket/websocket_client.rs
@@ -59,6 +59,12 @@ const REQUEST_INITIAL_BYTES_LENGTH: usize = 4;
#[cfg(not(feature = "vsr"))]
const RESPONSE_INITIAL_BYTES_LENGTH: usize = 8;
const NAME: &str = "WebSocket";
+/// Bound on how long a single VSR reply read may block. The connection is
+/// lockstep and the read runs in the caller's task while holding the stream
+/// lock, so an unanswered read (lost server reply) would wedge every later
+/// request on this client forever. On expiry the stream is dropped.
+#[cfg(feature = "vsr")]
+const RESPONSE_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
#[derive(Debug)]
pub struct WebSocketClient {
@@ -697,8 +703,23 @@ impl WebSocketClient {
// Old path did `vec![0; HEADER_SIZE]` + `vec![0; body_size]`
// (two zero-fills) + `BytesMut::with_capacity(response_size)` +
// two `put_slice` memcopies per reply.
+ // One deadline spans both the header and body reads so a reply
+ // that delivers a header then stalls cannot wait up to 2x the
+ // timeout. On expiry drop the stream: the read runs inline here
+ // (no spawned task to cancel), so without an explicit drop a late
+ // reply would desync framing for the next request.
+ let response_deadline = tokio::time::Instant::now() + RESPONSE_READ_TIMEOUT;
let mut response_header = [0u8; iggy_binary_protocol::HEADER_SIZE];
- stream.read(&mut response_header).await?;
+ let header_read =
+ tokio::time::timeout_at(response_deadline, stream.read(&mut response_header)).await;
+ let Ok(header_read) = header_read else {
+ error!(
+ "Timed out after {RESPONSE_READ_TIMEOUT:?} waiting for {NAME} VSR response header for request with code: {code}"
+ );
+ *stream_guard = None;
+ return Err(IggyError::Disconnected);
+ };
+ header_read?;
let response_size = crate::vsr::response_size(&response_header)?;
let body_size = response_size - iggy_binary_protocol::HEADER_SIZE;
@@ -707,7 +728,16 @@ impl WebSocketClient {
// zero-fill prerequisite; we still allocate `body_size` but
// skip the header concatenation.
let mut body = vec![0u8; body_size];
- stream.read(&mut body).await?;
+ let body_read =
+ tokio::time::timeout_at(response_deadline, stream.read(&mut body)).await;
+ let Ok(body_read) = body_read else {
+ error!(
+ "Timed out after {RESPONSE_READ_TIMEOUT:?} waiting for {NAME} VSR response body for request with code: {code}"
+ );
+ *stream_guard = None;
+ return Err(IggyError::Disconnected);
+ };
+ body_read?;
Bytes::from(body)
} else {
Bytes::new()
diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs
index 0e5b76f98..ba5fa7a85 100644
--- a/core/server-ng/src/dispatch.rs
+++ b/core/server-ng/src/dispatch.rs
@@ -893,18 +893,11 @@ async fn handle_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.
+#[allow(clippy::future_not_send)]
async fn send_empty_partition_reply(
shard: &Rc<ServerNgShard>,
transport_client_id: u128,
@@ -926,6 +919,14 @@ async fn send_empty_partition_reply(
}
}
+/// 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)]
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);
@@ -942,8 +943,10 @@ async fn wait_for_partition_routable(shard: &Rc<ServerNgShard>, namespace: IggyN
}
// 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
- // owner's "partition not initialized" guard.
+ // until the owner answers, so the write below normally clears the
+ // owner's "partition not initialized" guard. Not a hard guarantee: the
+ // partition can de-materialise between this probe and the dispatch, but
+ // the park/tombstone path re-checks and the client retries.
while std::time::Instant::now() < deadline {
match shard
.partition_read(
diff --git a/core/server-ng/src/responses.rs b/core/server-ng/src/responses.rs
index 7a638f816..e1102ad49 100644
--- a/core/server-ng/src/responses.rs
+++ b/core/server-ng/src/responses.rs
@@ -30,13 +30,13 @@ use bytes::Bytes;
use consensus::{MetadataHandle, VsrConsensus};
use iggy_binary_protocol::codes::{
GET_CLUSTER_METADATA_CODE, GET_STATS_CODE, GET_STREAM_CODE, GET_STREAMS_CODE, GET_TOPIC_CODE,
- GET_TOPICS_CODE, GET_USER_CODE, GET_USERS_CODE, POLL_MESSAGES_CODE,
+ GET_TOPICS_CODE, GET_USER_CODE, GET_USERS_CODE,
};
use iggy_binary_protocol::requests::consumer_offsets::{
DeleteConsumerOffset2Request, DeleteConsumerOffsetRequest, StoreConsumerOffset2Request,
StoreConsumerOffsetRequest,
};
-use iggy_binary_protocol::requests::messages::{PollMessagesRequest, SendMessagesHeader};
+use iggy_binary_protocol::requests::messages::SendMessagesHeader;
use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest;
use iggy_binary_protocol::requests::streams::{GetStreamRequest, GetStreamsRequest};
use iggy_binary_protocol::requests::topics::{GetTopicRequest, GetTopicsRequest};
@@ -291,13 +291,6 @@ pub(crate) fn build_non_replicated_response(
})
})
}
- POLL_MESSAGES_CODE => {
- let request =
- PollMessagesRequest::decode_from(body).map_err(|_| IggyError::InvalidCommand)?;
- Ok(NonReplicatedResponse::EmptyPolledMessages {
- partition_id: request.partition_id.unwrap_or(0),
- })
- }
_ => match iggy_binary_protocol::dispatch::lookup_command(code) {
Some(meta) if !meta.is_replicated() => Ok(NonReplicatedResponse::Empty),
Some(_) => Err(IggyError::FeatureUnavailable),
@@ -594,7 +587,6 @@ fn partition_response(
pub(crate) enum NonReplicatedResponse {
Empty,
- EmptyPolledMessages { partition_id: u32 },
Bytes(Bytes),
}
@@ -608,13 +600,6 @@ impl NonReplicatedResponse {
) -> Message<ReplyHeader> {
match self {
Self::Empty => build_empty_reply(request_header, client_id, session, commit),
- Self::EmptyPolledMessages { partition_id } => {
- build_reply_with_body(request_header, client_id, session, commit, 16, |body| {
- body[..4].copy_from_slice(&partition_id.to_le_bytes());
- body[4..12].copy_from_slice(&0u64.to_le_bytes());
- body[12..16].copy_from_slice(&0u32.to_le_bytes());
- })
- }
Self::Bytes(body) => {
build_reply_from_bytes(request_header, client_id, session, commit, &body)
}
@@ -814,7 +799,11 @@ pub(crate) fn build_polled_messages_body(
}
let offset = batch.base_offset + u64::from(offset_delta);
- let timestamp = batch.base_timestamp + u64::from(timestamp_delta);
+ // `base_timestamp` is the flat broker append time stamped once per
+ // batch; `timestamp_delta` is a per-message delta against the
+ // producer origin, so it only applies to `origin_timestamp`. Adding
+ // it to the broker base would mix two clocks.
+ let timestamp = batch.base_timestamp;
let origin_timestamp = batch.origin_timestamp + u64::from(timestamp_delta);
messages.extend_from_slice(checksum);
diff --git a/core/server_common/src/send_messages2.rs b/core/server_common/src/send_messages2.rs
index 05af70d84..71321c726 100644
--- a/core/server_common/src/send_messages2.rs
+++ b/core/server_common/src/send_messages2.rs
@@ -460,26 +460,6 @@ pub struct SendMessages2MessageView<'a> {
pub payload: &'a [u8],
}
-#[allow(dead_code)]
-impl SendMessages2MessageView<'_> {
- pub fn owned_message(&self, batch: &SendMessages2Header) -> IggyMessage2 {
- IggyMessage2 {
- header: IggyMessage2Header {
- checksum: self.header.checksum,
- id: self.header.id,
- offset: batch.base_offset + u64::from(self.header.offset_delta),
- timestamp: batch.base_timestamp,
- origin_timestamp: batch.origin_timestamp + u64::from(self.header.timestamp_delta),
- user_headers_length: self.header.user_headers_length,
- payload_length: self.header.payload_length,
- },
- payload: Bytes::copy_from_slice(self.payload),
- user_headers: (!self.user_headers.is_empty())
- .then(|| Bytes::copy_from_slice(self.user_headers)),
- }
- }
-}
-
#[allow(dead_code)]
pub struct SendMessages2Iterator<'a> {
blob: &'a [u8],
@@ -558,19 +538,16 @@ pub fn convert_request_message(
let request_header = *message.header();
let total_size = request_header.size as usize;
let body = &message.as_slice()[std::mem::size_of::<RequestHeader>()..total_size];
- if decode_request_slice(body).is_ok() {
+ if decode_batch_slice(body).is_ok() {
return Ok(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.
+/// Decode one batch slice (`[256B command header][blob]`), validating the
+/// batch checksum. The persisted segment-file record and the request slice
+/// share this layout, so both decode through here.
pub fn decode_batch_slice(body: &[u8]) -> Result<SendMessages2Ref<'_>, IggyError> {
- decode_request_slice(body)
-}
-
-fn decode_request_slice(body: &[u8]) -> Result<SendMessages2Ref<'_>, IggyError> {
if body.len() < COMMAND_HEADER_SIZE {
return Err(IggyError::InvalidCommand);
}
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index b6a682f77..a83fa36fb 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -1069,6 +1069,11 @@ where
// on data that is already gone.
let removed = partitions.remove(&namespace);
partitions.untombstone(&namespace);
+ // A topic created then deleted before its `InsertOwned`
+ // pass never drains parked frames the normal way; reclaim
+ // them here so they cannot leak across many create-delete
+ // races (the partition is gone, so the frames are moot).
+ self.discard_parked_partition_frames(namespace);
self.metrics.record_partition_removed();
confirmed_remove = true;
if removed.is_none() {
@@ -1081,6 +1086,7 @@ where
}
ReconcileOp::RemoveRouted { namespace } => {
self.shards_table.remove(&namespace);
+ self.discard_parked_partition_frames(namespace);
}
}
}
@@ -1156,6 +1162,25 @@ where
}
}
+ /// Drop parked frames for a namespace that will never materialise (it was
+ /// removed before its `ReconcileOp::InsertOwned`), so the pending entry is
+ /// reclaimed instead of leaking until process exit.
+ fn discard_parked_partition_frames(&self, namespace: IggyNamespace) {
+ if let Some(frames) = self
+ .pending_partition_frames
+ .borrow_mut()
+ .remove(&namespace)
+ && !frames.is_empty()
+ {
+ tracing::debug!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ count = frames.len(),
+ "discarding parked partition frames for removed namespace"
+ );
+ }
+ }
+
/// 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).