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/07/03 10:40:53 UTC
(iggy) branch master updated: fix(integration): pass remaining tests for server-ng (#3585)
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 7b2db334f fix(integration): pass remaining tests for server-ng (#3585)
7b2db334f is described below
commit 7b2db334f244df06805f2c76c0490a5c1a6ec7e5
Author: Grzegorz Koszyk <11...@users.noreply.github.com>
AuthorDate: Fri Jul 3 12:40:49 2026 +0200
fix(integration): pass remaining tests for server-ng (#3585)
Fixes remaining integration tests using server-ng binary.
---
.typos.toml | 3 +
Cargo.lock | 1 +
core/ai/mcp/Cargo.toml | 1 +
core/binary_protocol/src/consensus/operation.rs | 23 +
core/common/src/error/iggy_error.rs | 2 +
core/configs/src/server_config/sharding.rs | 2 +-
core/consensus/src/impls.rs | 23 +-
core/consensus/src/metadata_helpers.rs | 24 +-
core/consensus/src/plane_helpers.rs | 64 +++
core/integration/src/harness/handle/server.rs | 27 +-
.../src/harness/orchestrator/builder.rs | 6 +-
.../src/harness/orchestrator/harness.rs | 83 ++-
.../verify_consumer_group_partition_assignment.rs | 17 +-
core/integration/tests/mod.rs | 7 +-
core/integration/tests/sdk/mcp_parity.rs | 613 +++++++++++++++++++++
core/integration/tests/sdk/mod.rs | 3 +-
core/integration/tests/sdk/raw.rs | 2 +-
core/integration/tests/server/cg_vsr.rs | 14 +-
.../tests/server/concurrent_addition.rs | 12 +-
core/integration/tests/server/general.rs | 24 +-
core/integration/tests/server/mod.rs | 3 +-
core/integration/tests/server/purge_delete.rs | 22 +-
...onsumer_group_duplicate_name_create_scenario.rs | 11 +-
core/integration/tests/server/scenarios/mod.rs | 13 -
.../scenarios/segment_rotation_race_scenario.rs | 23 +
.../scenarios/stream_size_validation_scenario.rs | 86 ++-
.../tests/server/scenarios/system_scenario.rs | 158 ++++--
core/integration/tests/server/specific.rs | 16 +-
core/message_bus/Cargo.toml | 1 +
core/message_bus/src/client_listener/mod.rs | 9 +-
core/message_bus/src/transports/quic.rs | 186 +++++--
core/metadata/src/impls/metadata.rs | 261 ++++++++-
core/metadata/src/stm/consumer_group.rs | 3 +-
core/metadata/src/stm/stream.rs | 230 +++++++-
core/partitions/src/iggy_index_reader.rs | 117 ++++
core/partitions/src/iggy_partition.rs | 214 +++++--
core/partitions/src/lib.rs | 3 +
core/sdk/src/quic/quic_client.rs | 160 ++++--
core/sdk/src/tcp/tcp_client.rs | 189 ++++---
core/sdk/src/vsr.rs | 82 ++-
core/sdk/src/websocket/websocket_client.rs | 173 +++---
core/server-ng/src/auth.rs | 40 +-
core/server-ng/src/bootstrap.rs | 305 +++++-----
core/server-ng/src/dispatch.rs | 68 ++-
core/server-ng/src/lib.rs | 1 +
core/server-ng/src/partition_helpers.rs | 5 +-
core/server-ng/src/partition_reconciler.rs | 21 +-
core/server-ng/src/responses.rs | 173 ++++--
core/server-ng/src/segment_recovery.rs | 306 ++++++++++
core/server_common/src/indexes_mut.rs | 10 +-
core/shard/src/lib.rs | 22 +-
51 files changed, 3105 insertions(+), 757 deletions(-)
diff --git a/.typos.toml b/.typos.toml
index 0dc2b4ae1..2cf8f2ff4 100644
--- a/.typos.toml
+++ b/.typos.toml
@@ -46,6 +46,9 @@ writeable = "writeable"
Paket = "Paket"
# Visual Studio dockerignore glob "[Pp]ublish.xml" tokenizes to "ublish"
ublish = "ublish"
+# Valid English prefix in hyphenated compounds ("mis-strides", "mis-decoded");
+# typos tokenizes on the hyphen and misreads the prefix as "miss"/"mist"
+mis = "mis"
# Exclude auto-generated/non-editable files from typos check
[files]
diff --git a/Cargo.lock b/Cargo.lock
index cf28d8b36..106aa2445 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -8287,6 +8287,7 @@ dependencies = [
"ahash 0.8.12",
"async-channel",
"blake3",
+ "bytemuck",
"bytes",
"compio",
"compio-quic",
diff --git a/core/ai/mcp/Cargo.toml b/core/ai/mcp/Cargo.toml
index 9cc785d27..8f4e55fbe 100644
--- a/core/ai/mcp/Cargo.toml
+++ b/core/ai/mcp/Cargo.toml
@@ -29,6 +29,7 @@ publish = false
[features]
systemd = ["dep:sd-notify", "dep:tokio-util"]
+vsr = ["iggy/vsr"]
[dependencies]
axum = { workspace = true }
diff --git a/core/binary_protocol/src/consensus/operation.rs b/core/binary_protocol/src/consensus/operation.rs
index e171095ea..954623132 100644
--- a/core/binary_protocol/src/consensus/operation.rs
+++ b/core/binary_protocol/src/consensus/operation.rs
@@ -155,6 +155,29 @@ impl Operation {
)
}
+ /// Whether a committed/rejected reply for this operation carries the
+ /// result section (`[count: u32]` then `count` x `{index, result}`) ahead
+ /// of the payload. This is the single source of truth the encode sites
+ /// (`ApplyReply::write_reply_body`, the partition plane's
+ /// `committed_reply_body`, `build_result_rejection_reply` callers) and the
+ /// SDK decode carve-out (`split_metadata_result`) must agree on; a drifted
+ /// site decodes a rejection as `Ok` or misreads a payload as the count.
+ /// Metadata ops are all result-framed; on the partition plane only the
+ /// consumer-offset ops are (they reject with typed errors at admission).
+ /// `Register` is result-framed only when non-empty -- that nuance stays in
+ /// the SDK, the one place that sees Register replies.
+ #[must_use]
+ pub const fn is_result_framed(&self) -> bool {
+ self.is_metadata()
+ || matches!(
+ self,
+ Self::StoreConsumerOffset
+ | Self::StoreConsumerOffset2
+ | Self::DeleteConsumerOffset
+ | Self::DeleteConsumerOffset2
+ )
+ }
+
/// Data-plane operations routed to the shard owning the partition.
#[must_use]
#[inline]
diff --git a/core/common/src/error/iggy_error.rs b/core/common/src/error/iggy_error.rs
index 22f1308c8..4bae28138 100644
--- a/core/common/src/error/iggy_error.rs
+++ b/core/common/src/error/iggy_error.rs
@@ -120,6 +120,8 @@ pub enum IggyError {
UsersLimitReached = 55,
#[error("Invalid personal access token expiry")]
InvalidPersonalAccessTokenExpiry = 56,
+ #[error("Request transiently not committed; retry")]
+ TransientNotCommitted = 57,
#[error("Not connected")]
NotConnected = 61,
#[error("Client shutdown")]
diff --git a/core/configs/src/server_config/sharding.rs b/core/configs/src/server_config/sharding.rs
index 7434f9167..ec332462b 100644
--- a/core/configs/src/server_config/sharding.rs
+++ b/core/configs/src/server_config/sharding.rs
@@ -25,7 +25,7 @@ use configs::ConfigEnv;
/// Default capacity of the per-shard inter-shard inbox channel. Sized
/// comfortably above the consensus working set, which is roughly
-/// `PIPELINE_PREPARE_QUEUE_MAX (= 8) * replica_count * directions`
+/// `PIPELINE_PREPARE_QUEUE_MAX (= 32) * replica_count * directions`
/// frames in flight per shard, without allowing a runaway producer to
/// eat unbounded memory. Tunable via `[system.sharding] inbox_capacity`
/// in TOML.
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index 0edcb2fee..53bb8aaac 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -85,7 +85,14 @@ impl Sequencer for LocalSequencer {
/// TODO The below numbers need to be added a consensus config
/// TODO understand how to configure these numbers.
/// Maximum number of prepares that can be in-flight in the pipeline.
-pub const PIPELINE_PREPARE_QUEUE_MAX: usize = 8;
+///
+/// Sized to absorb a synchronized client burst (e.g. the 20-way
+/// concurrent-creation race tests across TCP/QUIC/WebSocket) without
+/// `PipelineFull`-rejecting and disconnecting clients that cannot replay in
+/// time. At depth 8 the QUIC burst wedges the metadata consensus even in
+/// release. Stays well under the journal's `SLOT_COUNT` (1024) and the inbox
+/// capacity headroom.
+pub const PIPELINE_PREPARE_QUEUE_MAX: usize = 32;
/// Max accepted-but-not-yet-prepared requests buffered behind a full
/// prepare queue. Beyond this, requests drop and the client retries.
@@ -777,6 +784,20 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
pub fn advance_commit_max(&self, commit: u64) {
if commit > self.commit_max.get() {
self.commit_max.set(commit);
+ // A prepare just committed. Re-arm the prepare-retransmit timer for
+ // the next-oldest pending prepare rather than letting it inherit the
+ // previous op's grown backoff: the push-site `start` is a no-op
+ // while ticking and nothing else clears `attempts`, so under
+ // sustained load the backoff ratchets up to 16x base and the tail
+ // op's retransmit fires too rarely to recover a lost backup ack,
+ // stalling commit. `start` (not `reset`) forces the timer ticking
+ // at the base interval - `reset` alone leaves `ticking` untouched,
+ // so a timer stopped earlier would be armed-but-dead and never
+ // fire. When nothing is pending the timer self-stops via the
+ // empty-pipeline branch in `handle_prepare_timeout`.
+ if self.sequencer.current_sequence() > commit {
+ self.timeouts.borrow_mut().start(TimeoutKind::Prepare);
+ }
}
assert!(self.commit_max.get() >= self.commit_min.get());
}
diff --git a/core/consensus/src/metadata_helpers.rs b/core/consensus/src/metadata_helpers.rs
index 2622d4bf3..b2a4ecfdc 100644
--- a/core/consensus/src/metadata_helpers.rs
+++ b/core/consensus/src/metadata_helpers.rs
@@ -46,8 +46,13 @@ pub enum PreflightOutcome {
/// Session gone (`NoSession`) or rotated past the retry (`SessionTooLow`):
/// the client must be told with an eviction frame.
Evict(EvictionReason),
- /// Absorbed with nothing to send: in-flight prepare, not-caught-up
- /// primary, stale/gap retry, or a client-bug newer session.
+ /// Transient: the request could not be committed *right now* but a replay of
+ /// the same `request_id` is expected to succeed (in-flight prepare,
+ /// not-caught-up primary). The caller sends a `TransientNotCommitted` reply
+ /// so the client replays immediately instead of waiting out its read-timeout.
+ NotReady,
+ /// Absorbed with nothing to send: stale/gap retry or a client-bug newer
+ /// session. Replaying the same `request_id` cannot help, so stay silent.
Drop,
}
@@ -77,9 +82,9 @@ where
tracing::debug!(
client_id,
request,
- "request_preflight: in-flight prepare, drop"
+ "request_preflight: in-flight prepare, not ready"
);
- return PreflightOutcome::Drop;
+ return PreflightOutcome::NotReady;
}
// Catch-up gate: stale ClientTable on a new primary could return `New`
@@ -95,9 +100,9 @@ where
is_syncing = consensus.is_syncing(),
commit_min = consensus.commit_min(),
commit_max = consensus.commit_max(),
- "request_preflight: not caught up, drop"
+ "request_preflight: not caught up, not ready"
);
- return PreflightOutcome::Drop;
+ return PreflightOutcome::NotReady;
}
let status = client_table
@@ -172,7 +177,12 @@ where
send_eviction_to_client(consensus, client_id, reason).await;
false
}
- PreflightOutcome::Drop => false,
+ // The wire-ingress plane has no per-request transport context to build a
+ // correlated `TransientNotCommitted` reply (that lives on the in-process
+ // home-shard path); stay silent here as before. NotReady and Drop both
+ // mean "do not dispatch"; the difference (explicit retry frame) only
+ // applies where the request header is in scope.
+ PreflightOutcome::NotReady | PreflightOutcome::Drop => false,
}
}
diff --git a/core/consensus/src/plane_helpers.rs b/core/consensus/src/plane_helpers.rs
index f40562cb3..32f027e37 100644
--- a/core/consensus/src/plane_helpers.rs
+++ b/core/consensus/src/plane_helpers.rs
@@ -334,6 +334,70 @@ where
.expect("reply buffer must contain a valid reply message")
}
+/// Builds a `Reply` carrying only a single-entry result section, no payload:
+/// the generic `[count=1][index=0][result=code]` rejection frame.
+///
+/// Two families of `code` ride this frame. `TransientNotCommitted` marks a
+/// request that could not be committed *right now* (not-caught-up / in-flight
+/// / pipeline-full / view-change cancel); the SDK decodes it and replays the
+/// same `request_id` immediately instead of waiting out its response
+/// read-timeout. Any other code is a TERMINAL rejection (e.g.
+/// `ConsumerOffsetNotFound` / `InvalidOffset` from partition-plane admission)
+/// that the SDK surfaces as the typed error.
+///
+/// Stamped from the request header (no prepare exists for a rejected op).
+/// `commit` is the primary's current commit position and is informational
+/// here: a rejection reply is sent, never cached in the `ClientTable`, so it
+/// is not subject to the `commit_reply` regression order.
+///
+/// # Panics
+/// If the constructed message buffer is not a valid reply.
+#[must_use]
+#[allow(clippy::cast_possible_truncation)]
+pub fn build_result_rejection_reply(
+ request_header: &RequestHeader,
+ commit: u64,
+ code: u32,
+) -> Message<ReplyHeader> {
+ // `[count: u32][index: u32][result: u32]`, the single-entry rejection shape
+ // of `ApplyReply::write_reply_body` (mirrored here to avoid a metadata dep).
+ const RESULT_BODY_LEN: usize = 12;
+ let header_size = std::mem::size_of::<ReplyHeader>();
+ let total_size = header_size + RESULT_BODY_LEN;
+ let mut buffer = bytes::BytesMut::zeroed(total_size);
+
+ let header = bytemuck::checked::try_from_bytes_mut::<ReplyHeader>(&mut buffer[..header_size])
+ .expect("zeroed bytes are valid");
+ *header = ReplyHeader {
+ cluster: request_header.cluster,
+ size: total_size as u32,
+ view: request_header.view,
+ release: request_header.release,
+ command: Command2::Reply,
+ replica: request_header.replica,
+ request_checksum: request_header.request_checksum,
+ client: request_header.client,
+ // Position-typed like the sibling builders (`build_reply_from_request`
+ // stamps `op: commit` too); inert on this path -- rejections are never
+ // cached -- but keeps the wire field convention for frame inspection.
+ op: commit,
+ commit,
+ timestamp: request_header.timestamp,
+ request: request_header.request,
+ operation: request_header.operation,
+ namespace: request_header.namespace,
+ ..Default::default()
+ };
+
+ let body = &mut buffer[header_size..];
+ body[0..4].copy_from_slice(&1u32.to_le_bytes());
+ body[4..8].copy_from_slice(&0u32.to_le_bytes());
+ body[8..12].copy_from_slice(&code.to_le_bytes());
+
+ Message::try_from(Owned::<4096>::copy_from_slice(buffer.as_ref()))
+ .expect("transient reply buffer must contain a valid reply message")
+}
+
/// Reply for fast paths that skip the VSR pipeline (e.g. `AckLevel::NoAck`).
///
/// Stamps `op` and `commit` with `commit_max` — monotonic, so
diff --git a/core/integration/src/harness/handle/server.rs b/core/integration/src/harness/handle/server.rs
index cdd04434f..5c96cc7f2 100644
--- a/core/integration/src/harness/handle/server.rs
+++ b/core/integration/src/harness/handle/server.rs
@@ -793,11 +793,28 @@ impl TestBinary for ServerHandle {
path: cert_dir.clone(),
source: e,
})?;
- generate_test_certificates(cert_dir.to_str().unwrap()).map_err(|e| {
- TestBinaryError::InvalidState {
- message: format!("Failed to generate TLS certificates: {e}"),
- }
- })?;
+ // One keypair per test dir (vsr only): the vsr harness spawns a
+ // multi-node cluster and every node's `start()` runs this block
+ // against the SHARED `certs/` dir. Regenerating would overwrite the
+ // PEM with a fresh keypair -- nodes started earlier then present a
+ // certificate signed by a different key than the one the client
+ // trusts (rcgen self-signed certs share the same subject DN), which
+ // rustls rejects as `BadSignature`. Generate only when absent so all
+ // nodes and clients share one keypair; this also keeps the
+ // certificate stable across a restart. The legacy single-node
+ // harness keeps its regenerate-per-start behavior.
+ #[cfg(feature = "vsr")]
+ let should_generate = !(cert_dir.join("test_cert.pem").exists()
+ && cert_dir.join("test_key.pem").exists());
+ #[cfg(not(feature = "vsr"))]
+ let should_generate = true;
+ if should_generate {
+ generate_test_certificates(cert_dir.to_str().unwrap()).map_err(|e| {
+ TestBinaryError::InvalidState {
+ message: format!("Failed to generate TLS certificates: {e}"),
+ }
+ })?;
+ }
self.generated_cert_dir = Some(cert_dir);
}
diff --git a/core/integration/src/harness/orchestrator/builder.rs b/core/integration/src/harness/orchestrator/builder.rs
index 81b0e1a44..aaa056cad 100644
--- a/core/integration/src/harness/orchestrator/builder.rs
+++ b/core/integration/src/harness/orchestrator/builder.rs
@@ -393,7 +393,7 @@ mod tests {
.build()
.unwrap();
- assert_eq!(harness.servers.len(), 1);
+ assert_eq!(harness.servers.len(), default_cluster_node_count());
assert!(!harness.started);
assert_eq!(harness.client_configs.len(), 1);
}
@@ -407,7 +407,7 @@ mod tests {
.build()
.unwrap();
- assert_eq!(harness.servers.len(), 1);
+ assert_eq!(harness.servers.len(), default_cluster_node_count());
assert!(harness.mcp().is_some());
}
@@ -452,6 +452,6 @@ mod tests {
.build()
.unwrap();
- assert_eq!(harness.servers.len(), 1);
+ assert_eq!(harness.servers.len(), default_cluster_node_count());
}
}
diff --git a/core/integration/src/harness/orchestrator/harness.rs b/core/integration/src/harness/orchestrator/harness.rs
index f715c52ad..7b136005f 100644
--- a/core/integration/src/harness/orchestrator/harness.rs
+++ b/core/integration/src/harness/orchestrator/harness.rs
@@ -35,6 +35,8 @@ use std::sync::Arc;
use std::time::{Duration, Instant};
#[cfg(feature = "vsr")]
use tokio::time::{sleep, timeout};
+#[cfg(feature = "vsr")]
+use tracing::warn;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
@@ -152,11 +154,56 @@ impl TestHarness {
self.jwks_server = Some(mock_server);
}
+ // Legacy single-server harness: plain start, no cluster readiness.
+ #[cfg(not(feature = "vsr"))]
for server in &mut self.servers {
server.start()?;
}
- self.wait_for_cluster_ready().await?;
+ // Cluster startup (vsr) can hit a transient replica-handshake blip that
+ // leaves the mesh incomplete (a peer link drops mid-handshake, so a
+ // node never reaches "all peers connected"). Rather than fail the whole
+ // test on a startup blip, retry spawn + mesh-readiness a few times,
+ // tearing down and respawning between attempts. `ServerHandle::start`
+ // truncates the captured stdout (`File::create`), so the readiness
+ // log-grep never matches a stale marker from a prior attempt.
+ #[cfg(feature = "vsr")]
+ {
+ const CLUSTER_STARTUP_ATTEMPTS: usize = 3;
+ for attempt in 1..=CLUSTER_STARTUP_ATTEMPTS {
+ let mut spawn_error = None;
+ for server in &mut self.servers {
+ if let Err(error) = server.start() {
+ spawn_error = Some(error);
+ break;
+ }
+ }
+ if let Some(error) = spawn_error {
+ // Mirror the not-ready arm below: never bail with earlier
+ // nodes of this attempt still running.
+ for server in &mut self.servers {
+ let _ = server.stop();
+ }
+ return Err(error);
+ }
+ match self.wait_for_cluster_ready().await {
+ Ok(()) => break,
+ Err(error) if attempt < CLUSTER_STARTUP_ATTEMPTS => {
+ warn!(
+ attempt,
+ max_attempts = CLUSTER_STARTUP_ATTEMPTS,
+ error = %error,
+ "cluster not ready; tearing down and respawning"
+ );
+ for server in &mut self.servers {
+ let _ = server.stop();
+ }
+ sleep(Duration::from_millis(500)).await;
+ }
+ Err(error) => return Err(error),
+ }
+ }
+ }
if let Some(seed_fn) = seed {
let client = self.tcp_root_client().await?;
@@ -172,13 +219,8 @@ impl TestHarness {
Ok(())
}
+ #[cfg(feature = "vsr")]
async fn wait_for_cluster_ready(&self) -> Result<(), TestBinaryError> {
- #[cfg(not(feature = "vsr"))]
- {
- Ok(())
- }
-
- #[cfg(feature = "vsr")]
{
if self.servers.len() <= 1 {
return Ok(());
@@ -203,9 +245,17 @@ impl TestHarness {
sleep(CLUSTER_READY_RETRY_INTERVAL).await;
}
if !self.servers.iter().all(ServerHandle::replica_mesh_complete) {
+ let per_node: Vec<String> = self
+ .servers
+ .iter()
+ .enumerate()
+ .map(|(i, s)| format!("node{i}={}", s.replica_mesh_complete()))
+ .collect();
return Err(TestBinaryError::InvalidState {
- message: "Timed out waiting for VSR replica mesh to form on all nodes"
- .to_string(),
+ message: format!(
+ "Timed out waiting for VSR replica mesh to form on all nodes [{}]",
+ per_node.join(", ")
+ ),
});
}
@@ -340,20 +390,15 @@ impl TestHarness {
/// Get the MCP handle from the primary server if configured.
///
- /// # Panics
- /// Panics if called on a cluster (multiple servers). Use `node(i).mcp()` instead.
+ /// MCP is attached to the primary (first) server for both single-server and
+ /// cluster setups (see the builder), so this returns the primary's handle in
+ /// either case.
pub fn mcp(&self) -> Option<&McpHandle> {
- assert!(
- self.servers.len() <= 1,
- "mcp() is only available for single-server setups. Use node(i).mcp() for clusters."
- );
self.servers.first().and_then(|s| s.mcp())
}
- /// Create an MCP client (convenience method).
- ///
- /// # Panics
- /// Panics if called on a cluster (multiple servers). Use `node(i).mcp()` instead.
+ /// Create an MCP client (convenience method). Targets the primary's MCP
+ /// handle, for both single-server and cluster setups.
pub async fn mcp_client(&self) -> Result<McpClient, TestBinaryError> {
self.mcp()
.ok_or(TestBinaryError::MissingMcp)?
diff --git a/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs b/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs
index 165367949..f5f378837 100644
--- a/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs
+++ b/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs
@@ -21,6 +21,7 @@ use std::collections::HashSet;
use std::str::FromStr;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::time::Instant;
use tokio::time::{Duration, sleep};
const STREAM_NAME: &str = "cg-partition-test-stream";
@@ -3249,11 +3250,23 @@ async fn should_not_complete_other_members_revocations_on_leave(harness: &TestHa
// 4. Consumer3 disconnects — triggers full rebalance (rebalance_members).
// Leave clears only consumer3's polled offsets, not consumer1's.
drop(client3);
- sleep(Duration::from_secs(3)).await;
// 5. After full rebalance, consumer1 and consumer2 split all partitions.
// The key invariant: no partition is assigned to two members.
- let cg = get_consumer_group(&root_client).await;
+ // Convergence is load-dependent: the dead connection is noticed by the
+ // heartbeat verifier (interval 2s, worst case ~2x), then the Leave
+ // commits and the reconciler rebalances -- a fixed sleep flakes under
+ // parallel-suite load. Poll until the membership and assignments
+ // converge, then make the terminal assertions.
+ let deadline = Instant::now() + Duration::from_secs(15);
+ let cg = loop {
+ let cg = get_consumer_group(&root_client).await;
+ let assigned: u32 = cg.members.iter().map(|m| m.partitions_count).sum();
+ if (cg.members_count == 2 && assigned == PARTITIONS_COUNT) || Instant::now() >= deadline {
+ break cg;
+ }
+ sleep(Duration::from_millis(200)).await;
+ };
assert_eq!(cg.members_count, 2);
assert_unique_partition_assignments(&cg);
let total: u32 = cg.members.iter().map(|m| m.partitions_count).sum();
diff --git a/core/integration/tests/mod.rs b/core/integration/tests/mod.rs
index c20bac18c..ee100b2d0 100644
--- a/core/integration/tests/mod.rs
+++ b/core/integration/tests/mod.rs
@@ -34,7 +34,6 @@ use tracing_subscriber::{EnvFilter, fmt};
mod cli;
#[cfg(not(feature = "vsr"))]
mod cluster;
-#[cfg(not(feature = "vsr"))]
mod config_provider;
#[cfg(not(feature = "vsr"))]
mod connectors;
@@ -63,6 +62,12 @@ mod connectors;
// revocation. server-ng's rebalance is also covered by `server::cg_vsr` +
// `stale_client_consumer_group_scenario` (both green).
mod data_integrity;
+// TODO(vsr): un-gate once the metadata `messages_count` aggregation gap is
+// fixed (partition-plane counts aren't shared into the metadata topic/stream
+// stats under vsr -- root-caused: per-buffer left-right `Arc<TopicStats>`;
+// see `sdk::metadata_counts`) and the `delete_consumer_offset` hang is
+// addressed. Enablement scaffolding is done: the `iggy-mcp` crate has a `vsr`
+// feature and the harness `mcp()` accessor works for clusters.
#[cfg(not(feature = "vsr"))]
mod mcp;
mod sdk;
diff --git a/core/integration/tests/sdk/mcp_parity.rs b/core/integration/tests/sdk/mcp_parity.rs
new file mode 100644
index 000000000..cd27ba25a
--- /dev/null
+++ b/core/integration/tests/sdk/mcp_parity.rs
@@ -0,0 +1,613 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Plain-SDK reproduction of the `mcp::*` suite, without the MCP server. Each
+//! MCP tool maps 1:1 to an SDK client method, so these tests exercise the same
+//! server-ng behaviour under the `vsr` feature over TCP, without the MCP/HTTP
+//! environment overhead. Assertions mirror the MCP counterparts.
+
+#![cfg(feature = "vsr")]
+
+use iggy::prelude::*;
+use integration::{
+ harness::{TestHarness, seeds},
+ iggy_harness,
+};
+
+/// The consumer under which `seeds::mcp_standard` stores its offset (matches the
+/// MCP server's default consumer name).
+fn seed_consumer() -> Consumer {
+ Consumer::new(Identifier::named(seeds::names::CONSUMER).unwrap())
+}
+
+async fn root_client(harness: &TestHarness) -> IggyClient {
+ let client = harness.new_client().await.unwrap();
+ client
+ .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+ .await
+ .unwrap();
+ client
+}
+
+// mcp::should_handle_ping
+#[iggy_harness(test_client_transport = [Tcp])]
+async fn should_handle_ping(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ client.ping().await.unwrap();
+}
+
+// mcp::should_return_cluster_metadata
+// TODO: MCP polls `get_cluster_metadata` via a dedicated tool; there is no
+// equivalent single SDK client method to assert cluster name + node count, so
+// this case is intentionally not ported here.
+
+// mcp::should_return_list_of_streams
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_return_list_of_streams(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let streams = client.get_streams().await.unwrap();
+
+ assert_eq!(streams.len(), 1);
+ assert_eq!(streams[0].name, seeds::names::STREAM);
+ assert_eq!(streams[0].topics_count, 1);
+}
+
+// mcp::should_return_stream_details
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_return_stream_details(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let stream = client
+ .get_stream(&stream_id)
+ .await
+ .unwrap()
+ .expect("stream exists");
+
+ assert_eq!(stream.name, seeds::names::STREAM);
+ assert_eq!(stream.topics_count, 1);
+ assert_eq!(stream.messages_count, 1);
+}
+
+// mcp::should_create_stream
+#[iggy_harness(test_client_transport = [Tcp])]
+async fn should_create_stream(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let name = "new_stream";
+ let stream = client.create_stream(name).await.unwrap();
+
+ assert_eq!(stream.name, name);
+ assert_eq!(stream.topics_count, 0);
+ assert_eq!(stream.messages_count, 0);
+}
+
+// mcp::should_update_stream
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_update_stream(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ client
+ .update_stream(&stream_id, "updated_stream")
+ .await
+ .unwrap();
+}
+
+// mcp::should_delete_stream
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_delete_stream(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ client.delete_stream(&stream_id).await.unwrap();
+}
+
+// mcp::should_purge_stream
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_purge_stream(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ client.purge_stream(&stream_id).await.unwrap();
+}
+
+// mcp::should_return_list_of_topics
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_return_list_of_topics(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let topics = client.get_topics(&stream_id).await.unwrap();
+
+ assert_eq!(topics.len(), 1);
+ assert_eq!(topics[0].name, seeds::names::TOPIC);
+ assert_eq!(topics[0].partitions_count, 1);
+ assert_eq!(topics[0].messages_count, 1);
+}
+
+// mcp::should_return_topic_details
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_return_topic_details(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
+ let topic = client
+ .get_topic(&stream_id, &topic_id)
+ .await
+ .unwrap()
+ .expect("topic exists");
+
+ assert_eq!(topic.id, 0);
+ assert_eq!(topic.name, seeds::names::TOPIC);
+ assert_eq!(topic.messages_count, 1);
+}
+
+// mcp::should_create_topic
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_create_topic(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let name = "new_topic";
+ let topic = client
+ .create_topic(
+ &stream_id,
+ name,
+ 1,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::ServerDefault,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .unwrap();
+
+ assert_eq!(topic.id, 1);
+ assert_eq!(topic.name, name);
+ assert_eq!(topic.partitions_count, 1);
+ assert_eq!(topic.messages_count, 0);
+}
+
+// mcp::should_update_topic
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_update_topic(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
+ client
+ .update_topic(
+ &stream_id,
+ &topic_id,
+ "updated_topic",
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::ServerDefault,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .unwrap();
+}
+
+// mcp::should_delete_topic
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_delete_topic(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
+ client.delete_topic(&stream_id, &topic_id).await.unwrap();
+}
+
+// mcp::should_purge_topic
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_purge_topic(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
+ client.purge_topic(&stream_id, &topic_id).await.unwrap();
+}
+
+// mcp::should_create_partitions
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_create_partitions(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
+ client
+ .create_partitions(&stream_id, &topic_id, 3)
+ .await
+ .unwrap();
+}
+
+// mcp::should_delete_partitions
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_delete_partitions(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
+ client
+ .delete_partitions(&stream_id, &topic_id, 1)
+ .await
+ .unwrap();
+}
+
+// mcp::should_delete_segments
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_delete_segments(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
+ client
+ .delete_segments(&stream_id, &topic_id, 0, 1)
+ .await
+ .unwrap();
+}
+
+// mcp::should_poll_messages
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_poll_messages(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
+ let messages = client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(0),
+ &seed_consumer(),
+ &PollingStrategy::offset(0),
+ 10,
+ false,
+ )
+ .await
+ .unwrap();
+
+ assert_eq!(messages.messages.len(), 1);
+ assert_eq!(messages.messages[0].header.offset, 0);
+ let payload = messages.messages[0]
+ .payload_as_string()
+ .expect("Failed to parse payload");
+ assert_eq!(payload, seeds::names::MESSAGE_PAYLOAD);
+}
+
+// mcp::should_send_messages
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_send_messages(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
+ let mut messages = vec![
+ IggyMessage::builder()
+ .payload("test".into())
+ .build()
+ .unwrap(),
+ ];
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .unwrap();
+}
+
+// mcp::should_return_stats
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_return_stats(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stats = client.get_stats().await.unwrap();
+
+ assert!(!stats.hostname.is_empty());
+ assert_eq!(stats.messages_count, 1);
+}
+
+// mcp::should_return_me
+#[iggy_harness(test_client_transport = [Tcp])]
+async fn should_return_me(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let me = client.get_me().await.unwrap();
+
+ assert!(me.client_id > 0);
+}
+
+// mcp::should_return_clients
+#[iggy_harness(test_client_transport = [Tcp])]
+async fn should_return_clients(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let clients = client.get_clients().await.unwrap();
+
+ assert!(!clients.is_empty());
+}
+
+// mcp::should_handle_snapshot
+// The diagnostic `snapshot` op (heap/config/process dump) is not implemented in
+// server-ng; it falls through to the non-replicated catch-all and returns an
+// empty body. Out of scope for metadata/stats parity -- un-ignore if server-ng
+// grows the snapshot subsystem.
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+#[ignore = "vsr: server-ng does not implement the diagnostic snapshot op"]
+async fn should_handle_snapshot(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let snapshot = client
+ .snapshot(
+ SnapshotCompression::default(),
+ SystemSnapshotType::all_snapshot_types(),
+ )
+ .await
+ .unwrap();
+
+ assert!(!snapshot.0.is_empty());
+}
+
+// mcp::should_return_consumer_groups
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_return_consumer_groups(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
+ let groups = client
+ .get_consumer_groups(&stream_id, &topic_id)
+ .await
+ .unwrap();
+
+ assert!(!groups.is_empty());
+}
+
+// mcp::should_return_consumer_group_details
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_return_consumer_group_details(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
+ let group_id = Identifier::named(seeds::names::CONSUMER_GROUP).unwrap();
+ let group = client
+ .get_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ .unwrap()
+ .expect("consumer group exists");
+
+ assert_eq!(group.name, seeds::names::CONSUMER_GROUP);
+ assert_eq!(group.partitions_count, 1);
+ assert_eq!(group.members_count, 0);
+ assert!(group.members.is_empty());
+}
+
+// mcp::should_create_consumer_group
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_create_consumer_group(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
+ let name = "new_group";
+ let group = client
+ .create_consumer_group(&stream_id, &topic_id, name)
+ .await
+ .unwrap();
+
+ assert_eq!(group.name, name);
+ assert_eq!(group.partitions_count, 1);
+ assert_eq!(group.members_count, 0);
+ assert!(group.members.is_empty());
+}
+
+// mcp::should_delete_consumer_group
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_delete_consumer_group(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
+ let group_id = Identifier::named(seeds::names::CONSUMER_GROUP).unwrap();
+ client
+ .delete_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ .unwrap();
+}
+
+// mcp::should_return_consumer_offset
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_return_consumer_offset(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
+ let offset = client
+ .get_consumer_offset(&seed_consumer(), &stream_id, &topic_id, Some(0))
+ .await
+ .unwrap()
+ .expect("Expected consumer offset");
+
+ assert_eq!(offset.partition_id, 0);
+ assert_eq!(offset.stored_offset, 0);
+ assert_eq!(offset.current_offset, 0);
+}
+
+// mcp::should_store_consumer_offset
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_store_consumer_offset(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
+ client
+ .store_consumer_offset(&seed_consumer(), &stream_id, &topic_id, Some(0), 0)
+ .await
+ .unwrap();
+}
+
+// mcp::should_delete_consumer_offset
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_delete_consumer_offset(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
+ client
+ .delete_consumer_offset(&seed_consumer(), &stream_id, &topic_id, Some(0))
+ .await
+ .unwrap();
+}
+
+// Regression (no MCP counterpart): deleting an offset that was never stored
+// must return a terminal `ConsumerOffsetNotFound`, NOT hang. The primary rejects
+// at admission with an explicit error reply instead of dropping the request
+// (which left the SDK replaying until its read-timeout).
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn delete_of_missing_consumer_offset_returns_not_found(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let stream_id = Identifier::named(seeds::names::STREAM).unwrap();
+ let topic_id = Identifier::named(seeds::names::TOPIC).unwrap();
+ let never_stored = Consumer::new(Identifier::named("never_stored").unwrap());
+ let result = client
+ .delete_consumer_offset(&never_stored, &stream_id, &topic_id, Some(0))
+ .await;
+ assert!(
+ matches!(result, Err(IggyError::ConsumerOffsetNotFound(_))),
+ "expected ConsumerOffsetNotFound, got {result:?}"
+ );
+}
+
+// mcp::should_return_personal_access_tokens
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_return_personal_access_tokens(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let tokens = client.get_personal_access_tokens().await.unwrap();
+
+ assert_eq!(tokens.len(), 1);
+ assert_eq!(tokens[0].name, seeds::names::PERSONAL_ACCESS_TOKEN);
+}
+
+// mcp::should_create_personal_access_token
+#[iggy_harness(test_client_transport = [Tcp])]
+async fn should_create_personal_access_token(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let name = "test_token";
+ let token = client
+ .create_personal_access_token(name, PersonalAccessTokenExpiry::NeverExpire)
+ .await
+ .unwrap();
+
+ assert!(!token.token.is_empty());
+}
+
+// mcp::should_delete_personal_access_token
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_delete_personal_access_token(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ client
+ .delete_personal_access_token(seeds::names::PERSONAL_ACCESS_TOKEN)
+ .await
+ .unwrap();
+}
+
+// mcp::should_return_users
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_return_users(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let users = client.get_users().await.unwrap();
+
+ assert_eq!(users.len(), 2);
+}
+
+// mcp::should_return_user_details
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_return_user_details(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let user_id = Identifier::named(seeds::names::USER).unwrap();
+ let user = client
+ .get_user(&user_id)
+ .await
+ .unwrap()
+ .expect("user exists");
+
+ assert_eq!(user.username, seeds::names::USER);
+}
+
+// mcp::should_create_user
+#[iggy_harness(test_client_transport = [Tcp])]
+async fn should_create_user(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let username = "test-mcp-user";
+ let user = client
+ .create_user(username, "secret", UserStatus::Active, None)
+ .await
+ .unwrap();
+
+ assert_eq!(user.username, username);
+}
+
+// mcp::should_update_user
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_update_user(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let user_id = Identifier::named(seeds::names::USER).unwrap();
+ client
+ .update_user(&user_id, Some("updated-user"), Some(UserStatus::Inactive))
+ .await
+ .unwrap();
+}
+
+// mcp::should_delete_user
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_delete_user(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let user_id = Identifier::named(seeds::names::USER).unwrap();
+ client.delete_user(&user_id).await.unwrap();
+}
+
+// mcp::should_update_permissions
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_update_permissions(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let user_id = Identifier::named(seeds::names::USER).unwrap();
+
+ let mut topics = std::collections::BTreeMap::new();
+ topics.insert(
+ 1,
+ TopicPermissions {
+ manage_topic: true,
+ read_topic: true,
+ poll_messages: true,
+ send_messages: true,
+ },
+ );
+ let mut streams = std::collections::BTreeMap::new();
+ streams.insert(
+ 1,
+ StreamPermissions {
+ manage_stream: true,
+ manage_topics: true,
+ topics: Some(topics),
+ ..Default::default()
+ },
+ );
+ let permissions = Permissions {
+ global: GlobalPermissions {
+ manage_servers: true,
+ read_users: true,
+ ..Default::default()
+ },
+ streams: Some(streams),
+ };
+
+ client
+ .update_permissions(&user_id, Some(permissions))
+ .await
+ .unwrap();
+}
+
+// mcp::should_change_password
+#[iggy_harness(test_client_transport = [Tcp], seed = seeds::mcp_standard)]
+async fn should_change_password(harness: &TestHarness) {
+ let client = root_client(harness).await;
+ let user_id = Identifier::named(seeds::names::USER).unwrap();
+ client
+ .change_password(&user_id, seeds::names::USER_PASSWORD, "new_secret")
+ .await
+ .unwrap();
+}
diff --git a/core/integration/tests/sdk/mod.rs b/core/integration/tests/sdk/mod.rs
index 2802a9f35..70b102c46 100644
--- a/core/integration/tests/sdk/mod.rs
+++ b/core/integration/tests/sdk/mod.rs
@@ -17,8 +17,9 @@
mod hello_world;
#[cfg(feature = "vsr")]
+mod mcp_parity;
+#[cfg(feature = "vsr")]
mod messages;
-#[cfg(not(feature = "vsr"))]
mod producer;
#[cfg(feature = "vsr")]
mod protocol_version;
diff --git a/core/integration/tests/sdk/raw.rs b/core/integration/tests/sdk/raw.rs
index f507603ac..cffeec954 100644
--- a/core/integration/tests/sdk/raw.rs
+++ b/core/integration/tests/sdk/raw.rs
@@ -32,7 +32,7 @@ async fn given_authenticated_client_when_sending_raw_request_should_round_trip(
}
#[cfg(feature = "vsr")]
-#[iggy_harness(test_client_transport = [Tcp, WebSocket])]
+#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic])]
async fn given_authenticated_client_when_sending_raw_request_should_round_trip(
harness: &TestHarness,
) {
diff --git a/core/integration/tests/server/cg_vsr.rs b/core/integration/tests/server/cg_vsr.rs
index 2448b41e1..ecf595a89 100644
--- a/core/integration/tests/server/cg_vsr.rs
+++ b/core/integration/tests/server/cg_vsr.rs
@@ -29,7 +29,7 @@ use crate::server::scenarios::{
use integration::iggy_harness;
#[iggy_harness(
- test_client_transport = [Tcp],
+ test_client_transport = [Tcp, WebSocket, Quic],
server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
)]
async fn duplicate_name_create_preserves_live_group(harness: &TestHarness) {
@@ -37,7 +37,7 @@ async fn duplicate_name_create_preserves_live_group(harness: &TestHarness) {
}
#[iggy_harness(
- test_client_transport = [Tcp],
+ test_client_transport = [Tcp, WebSocket, Quic],
server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
)]
async fn join(harness: &TestHarness) {
@@ -45,7 +45,7 @@ async fn join(harness: &TestHarness) {
}
#[iggy_harness(
- test_client_transport = [Tcp],
+ test_client_transport = [Tcp, WebSocket, Quic],
server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
)]
async fn single_client(harness: &TestHarness) {
@@ -53,7 +53,7 @@ async fn single_client(harness: &TestHarness) {
}
#[iggy_harness(
- test_client_transport = [Tcp],
+ test_client_transport = [Tcp, WebSocket, Quic],
server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
)]
async fn multiple_clients(harness: &TestHarness) {
@@ -61,7 +61,7 @@ async fn multiple_clients(harness: &TestHarness) {
}
#[iggy_harness(
- test_client_transport = [Tcp],
+ test_client_transport = [Tcp, WebSocket, Quic],
server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
)]
async fn offset_cleanup(harness: &TestHarness) {
@@ -69,7 +69,7 @@ async fn offset_cleanup(harness: &TestHarness) {
}
#[iggy_harness(
- test_client_transport = [Tcp],
+ test_client_transport = [Tcp, WebSocket, Quic],
server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
)]
async fn auto_commit_reconnection(harness: &TestHarness) {
@@ -77,7 +77,7 @@ async fn auto_commit_reconnection(harness: &TestHarness) {
}
#[iggy_harness(
- test_client_transport = [Tcp],
+ test_client_transport = [Tcp, WebSocket, Quic],
server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
)]
async fn new_messages_after_restart(harness: &TestHarness) {
diff --git a/core/integration/tests/server/concurrent_addition.rs b/core/integration/tests/server/concurrent_addition.rs
index ff1a4347e..e590078f1 100644
--- a/core/integration/tests/server/concurrent_addition.rs
+++ b/core/integration/tests/server/concurrent_addition.rs
@@ -37,12 +37,19 @@ use test_case::test_matrix;
quic.max_idle_timeout = "500s",
quic.keep_alive_interval = "15s"
))]
-#[test_matrix(
+// vsr excludes HTTP only (server-ng exposes no HTTP listener).
+#[cfg_attr(not(feature = "vsr"), test_matrix(
[tcp(), http(), quic(), websocket()],
[user(), stream(), topic(), partition(), consumer_group()],
[hot(), cold()],
[barrier_on(), barrier_off()]
-)]
+))]
+#[cfg_attr(feature = "vsr", test_matrix(
+ [tcp(), quic(), websocket()],
+ [user(), stream(), topic(), partition(), consumer_group()],
+ [hot(), cold()],
+ [barrier_on(), barrier_off()]
+))]
async fn matrix(
harness: &TestHarness,
transport: TransportProtocol,
@@ -57,6 +64,7 @@ fn tcp() -> TransportProtocol {
TransportProtocol::Tcp
}
+#[cfg(not(feature = "vsr"))]
fn http() -> TransportProtocol {
TransportProtocol::Http
}
diff --git a/core/integration/tests/server/general.rs b/core/integration/tests/server/general.rs
index 9f6a206de..eb1a88d57 100644
--- a/core/integration/tests/server/general.rs
+++ b/core/integration/tests/server/general.rs
@@ -17,12 +17,12 @@
use crate::server::scenarios::{
authentication_scenario, consumer_timestamp_polling_scenario, create_message_payload,
- message_headers_scenario,
+ invalid_consumer_offset_scenario, message_headers_scenario, stream_size_validation_scenario,
+ system_scenario,
};
#[cfg(not(feature = "vsr"))]
use crate::server::scenarios::{
- bench_scenario, invalid_consumer_offset_scenario, permissions_scenario, snapshot_scenario,
- stream_size_validation_scenario, system_scenario, user_scenario,
+ bench_scenario, permissions_scenario, snapshot_scenario, user_scenario,
};
use integration::iggy_harness;
@@ -40,11 +40,6 @@ async fn authentication(harness: &TestHarness) {
authentication_scenario::run(harness).await;
}
-// Blocked under vsr: asserts on responses server-ng still stubs --
-// `MaxTopicSize::ServerDefault` is echoed instead of resolved against
-// server config, and stats/cluster-metadata fields are hardcoded zeros.
-// Re-enable as the response wiring lands.
-#[cfg(not(feature = "vsr"))]
#[iggy_harness(
test_client_transport = [Tcp, Http, Quic, WebSocket],
server(
@@ -126,9 +121,6 @@ async fn create_message_payload_scenario(harness: &TestHarness) {
create_message_payload::run(harness).await;
}
-// Blocked under vsr: stream/topic size accounting is not surfaced into
-// the get_stream/get_topic responses yet (sizes report 0).
-#[cfg(not(feature = "vsr"))]
#[iggy_harness(
test_client_transport = [Tcp, Http, Quic, WebSocket],
server(
@@ -188,12 +180,10 @@ async fn snapshot(harness: &TestHarness) {
snapshot_scenario::run(harness).await;
}
-// Blocked under vsr (unsolved): expects typed errors (`InvalidOffset`)
-// for invalid offset stores, but the partition plane neither validates
-// stored offsets nor has a wire vehicle for per-request errors (the vsr
-// Reply carries no status; Eviction is session-terminal). Needs an
-// error-reply design on the partition plane.
-#[cfg(not(feature = "vsr"))]
+// Under vsr the partition plane validates the stored offset at admission and
+// replies with a terminal, result-framed `InvalidOffset` (see
+// `store_consumer_offset` handling in `iggy_partition::on_request`). HTTP is
+// auto-dropped by the harness macro, so this runs over [Tcp, Quic, WebSocket].
#[iggy_harness(
test_client_transport = [Tcp, Http, Quic, WebSocket],
server(
diff --git a/core/integration/tests/server/mod.rs b/core/integration/tests/server/mod.rs
index 53e8c70a4..eaed6e591 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -25,8 +25,7 @@ mod cg;
#[cfg(feature = "vsr")]
mod cg_vsr;
// 80-case race matrix with hardcoded HTTP variants (test_matrix bypasses
-// the harness transport filter); revisit under vsr once basics are green.
-#[cfg(not(feature = "vsr"))]
+// the harness transport filter).
mod concurrent_addition;
mod general;
// The per-shard segment cleaner deletes expired / oversize segments from disk
diff --git a/core/integration/tests/server/purge_delete.rs b/core/integration/tests/server/purge_delete.rs
index 9888fb97c..d190de21f 100644
--- a/core/integration/tests/server/purge_delete.rs
+++ b/core/integration/tests/server/purge_delete.rs
@@ -44,10 +44,12 @@ async fn should_delete_segments_and_validate_filesystem(
partition.messages_required_to_save = "1",
partition.enforce_fsync = "true",
))]
-// server-ng restarts the cluster on restart_on, which currently fails in
-// bootstrap independently of segment deletion: an `Index data must be exactly
-// 16 bytes` recovery panic (cache_indexes all/none) and a replica-port rebind
-// race. Run only the no-restart matrix under vsr until those are fixed.
+// Recovery on restart is fixed -- server-ng reads its own 24-byte segment index
+// via `segment_recovery` (no more "Index data must be exactly 16 bytes" panic;
+// the server restarts + re-meshes cleanly). restart_on stays vsr-gated on a
+// CLIENT blocker: the SDK `ConsensusSession` cannot reconnect after restart --
+// `register_request_id already called` (core/sdk/src/session.rs:121), the same
+// reconnect/re-register lifecycle `user_scenario` hits. Re-enable once that lands.
#[cfg_attr(not(feature = "vsr"), test_matrix([restart_off(), restart_on()]))]
#[cfg_attr(feature = "vsr", test_matrix([restart_off()]))]
async fn should_delete_segments_without_consumers(harness: &mut TestHarness, restart_server: bool) {
@@ -97,11 +99,11 @@ async fn should_block_deletion_until_all_consumers_pass_segment(
partition.messages_required_to_save = "1",
partition.enforce_fsync = "true",
))]
-// restart_on is vsr-gated: cluster restart hits a pre-existing server-ng
-// bootstrap failure (legacy 16-byte index reader vs server-ng 24-byte writer),
-// unrelated to purge. The scenario asserts the exact [0, 7, 14, 21] layout only
-// on the legacy path; under vsr it verifies the framing-agnostic purge outcome
-// (offsets cleared, files deleted, partition reset to a single segment at 0).
+// The scenario asserts the exact [0, 7, 14, 21] layout only on the legacy path;
+// under vsr it verifies the framing-agnostic purge outcome (offsets cleared,
+// files deleted, partition reset to a single segment at offset 0). restart_on
+// stays vsr-gated on the SDK reconnect-after-restart lifecycle (see
+// `should_delete_segments_without_consumers`).
#[cfg_attr(not(feature = "vsr"), test_matrix([restart_off(), restart_on()]))]
#[cfg_attr(feature = "vsr", test_matrix([restart_off()]))]
async fn should_purge_topic_and_clear_consumer_offsets(
@@ -115,6 +117,8 @@ fn restart_off() -> bool {
false
}
+// Only the `not(vsr)` matrix cell uses restart_on (restart is vsr-gated on the
+// SDK reconnect lifecycle above), so it is unused under the vsr feature.
#[cfg(not(feature = "vsr"))]
fn restart_on() -> bool {
true
diff --git a/core/integration/tests/server/scenarios/consumer_group_duplicate_name_create_scenario.rs b/core/integration/tests/server/scenarios/consumer_group_duplicate_name_create_scenario.rs
index fa1b928cb..71c310d5b 100644
--- a/core/integration/tests/server/scenarios/consumer_group_duplicate_name_create_scenario.rs
+++ b/core/integration/tests/server/scenarios/consumer_group_duplicate_name_create_scenario.rs
@@ -30,7 +30,9 @@ use iggy::prelude::Identifier;
use iggy::prelude::IggyExpiry;
use iggy::prelude::MaxTopicSize;
use iggy::prelude::{ConsumerGroupClient, StreamClient, TopicClient};
-use integration::harness::{TestHarness, assert_clean_system, create_user, login_user};
+use integration::harness::{
+ TestHarness, assert_clean_system, create_user, delete_user, login_user,
+};
pub async fn run(harness: &TestHarness) {
let system_client = harness
@@ -99,7 +101,12 @@ pub async fn run(harness: &TestHarness) {
"the surviving member must keep its partition assignment"
);
- cleanup(&system_client, true).await;
+ // This scenario creates only USERNAME_1, so delete just that user (the
+ // shared `cleanup(_, true)` deletes USERNAME_1/2/3 and would fail with
+ // UserNotFound on the two it never created); `cleanup(_, false)` drops the
+ // stream.
+ delete_user(&system_client, USERNAME_1).await;
+ cleanup(&system_client, false).await;
assert_clean_system(&system_client).await;
}
diff --git a/core/integration/tests/server/scenarios/mod.rs b/core/integration/tests/server/scenarios/mod.rs
index 7b02e107d..be99a9ad9 100644
--- a/core/integration/tests/server/scenarios/mod.rs
+++ b/core/integration/tests/server/scenarios/mod.rs
@@ -18,12 +18,7 @@
pub mod authentication_scenario;
#[cfg(not(feature = "vsr"))]
pub mod bench_scenario;
-// Concurrent produce+consume race regression: trips a metadata-plane
-// consensus race under vsr (`on_ack: committed prepare must be in journal`
-// panic on the primary); skip until the metadata races are fixed.
-#[cfg(not(feature = "vsr"))]
pub mod concurrent_produce_consume_scenario;
-#[cfg(not(feature = "vsr"))]
pub mod concurrent_scenario;
pub mod consumer_group_auto_commit_reconnection_scenario;
// Only consumed by `cg_vsr` (server-ng); legacy `cg.rs` has no equivalent.
@@ -40,7 +35,6 @@ pub mod create_message_payload;
pub mod cross_protocol_pat_scenario;
#[cfg(not(feature = "vsr"))]
pub mod encryption_scenario;
-#[cfg(not(feature = "vsr"))]
pub mod invalid_consumer_offset_scenario;
// Asserts server log-file rotation/archival policies; server-ng's file
// logger only captures bootstrap output (shard-thread logs never reach the
@@ -59,24 +53,19 @@ pub mod read_during_persistence_scenario;
pub mod reconnect_after_restart_scenario;
#[cfg(not(feature = "vsr"))]
pub mod restart_offset_skip_scenario;
-#[cfg(not(feature = "vsr"))]
pub mod segment_rotation_race_scenario;
pub mod single_message_per_batch_scenario;
#[cfg(not(feature = "vsr"))]
pub mod snapshot_scenario;
pub mod stale_client_consumer_group_scenario;
-#[cfg(not(feature = "vsr"))]
pub mod stream_size_validation_scenario;
#[cfg(feature = "vsr")]
pub mod stress_produce_consume_scenario;
-#[cfg(not(feature = "vsr"))]
pub mod system_scenario;
-#[cfg(not(feature = "vsr"))]
pub mod tcp_tls_scenario;
pub mod timestamp_scenario;
#[cfg(not(feature = "vsr"))]
pub mod user_scenario;
-#[cfg(not(feature = "vsr"))]
pub mod websocket_tls_scenario;
use iggy::prelude::*;
@@ -94,7 +83,6 @@ const CONSUMER_GROUP_NAME: &str = "test-consumer-group";
const USERNAME_1: &str = "user1";
const USERNAME_2: &str = "user2";
const USERNAME_3: &str = "user3";
-#[cfg(not(feature = "vsr"))]
const CONSUMER_KIND: ConsumerKind = ConsumerKind::Consumer;
const MESSAGES_COUNT: u32 = 1337;
@@ -165,7 +153,6 @@ async fn join_consumer_group(client: &IggyClient) {
.unwrap();
}
-#[cfg(not(feature = "vsr"))]
async fn leave_consumer_group(client: &IggyClient) {
client
.leave_consumer_group(
diff --git a/core/integration/tests/server/scenarios/segment_rotation_race_scenario.rs b/core/integration/tests/server/scenarios/segment_rotation_race_scenario.rs
index 6db1b65b0..6815ebac1 100644
--- a/core/integration/tests/server/scenarios/segment_rotation_race_scenario.rs
+++ b/core/integration/tests/server/scenarios/segment_rotation_race_scenario.rs
@@ -48,6 +48,15 @@ const MAX_ALLOWED_MEMORY_BYTES: u64 = 200 * 1024 * 1024;
/// Uses all available transports from the harness (TCP, HTTP, QUIC, WebSocket).
/// 2 producers are spawned per protocol, all writing to the same partition.
pub async fn run(harness: &TestHarness) {
+ // HTTP (REST) carries no VSR framing, so under vsr the race runs over the
+ // three VSR transports; the legacy path exercises all four.
+ #[cfg(feature = "vsr")]
+ let transports = [
+ TransportProtocol::Tcp,
+ TransportProtocol::Quic,
+ TransportProtocol::WebSocket,
+ ];
+ #[cfg(not(feature = "vsr"))]
let transports = [
TransportProtocol::Tcp,
TransportProtocol::Http,
@@ -70,6 +79,12 @@ pub async fn run(harness: &TestHarness) {
let server_pid = stats.process_id;
init_system(&admin_client, total_producers).await;
+ let memory_before_load = get_process_memory(server_pid);
+ println!(
+ "Server memory before load: {:.2} MB",
+ memory_before_load as f64 / 1024.0 / 1024.0
+ );
+
let stop_flag = Arc::new(AtomicBool::new(false));
let total_messages = Arc::new(AtomicU64::new(0));
let mut join_set = JoinSet::new();
@@ -127,6 +142,14 @@ pub async fn run(harness: &TestHarness) {
"Final server memory: {:.2} MB",
final_memory as f64 / 1024.0 / 1024.0
);
+ let final_stats = admin_client.get_stats().await.unwrap();
+ println!(
+ "Final segments_count: {} (memory per segment: {:.1} KiB)",
+ final_stats.segments_count,
+ (final_memory.saturating_sub(memory_before_load)) as f64
+ / 1024.0
+ / f64::from(final_stats.segments_count.max(1))
+ );
assert!(
final_memory < MAX_ALLOWED_MEMORY_BYTES,
"Memory leak detected! Server using {:.2} MB, max allowed is {:.2} MB",
diff --git a/core/integration/tests/server/scenarios/stream_size_validation_scenario.rs b/core/integration/tests/server/scenarios/stream_size_validation_scenario.rs
index 61806f4ee..ffa38a84a 100644
--- a/core/integration/tests/server/scenarios/stream_size_validation_scenario.rs
+++ b/core/integration/tests/server/scenarios/stream_size_validation_scenario.rs
@@ -20,15 +20,39 @@ use bytes::Bytes;
use iggy::prelude::*;
use integration::harness::{TestHarness, assert_clean_system, login_root};
use std::str::FromStr;
+use std::time::{Duration, Instant};
+use tokio::time::sleep;
+
+// The partition plane applies committed ops asynchronously on the owning
+// shard (sends fold into the shared stats at commit-apply; purge/delete
+// zero them when the reconciler drives the wipe), so a read racing that
+// window can see a pre-apply value. Retry until the expectation holds,
+// then make the terminal assertion for a real mismatch.
+const STATS_CONVERGENCE_TIMEOUT: Duration = Duration::from_secs(10);
+const STATS_RETRY_INTERVAL: Duration = Duration::from_millis(100);
const S1_NAME: &str = "test-stream-1";
const T1_NAME: &str = "test-topic-1";
const S2_NAME: &str = "test-stream-2";
const T2_NAME: &str = "test-topic-2";
const MESSAGE_PAYLOAD_SIZE_BYTES: u64 = 57;
+#[cfg(not(feature = "vsr"))]
const MSG_SIZE: u64 = IGGY_MESSAGE_HEADER_SIZE as u64 + MESSAGE_PAYLOAD_SIZE_BYTES; // number of bytes in a single message
const MSGS_COUNT: u64 = 117; // number of messages in a single topic after one pass of appending
+#[cfg(not(feature = "vsr"))]
const MSGS_SIZE: u64 = MSG_SIZE * MSGS_COUNT; // number of bytes in a single topic after one pass of appending
+// server-ng accounts the actual on-disk batch framing: one 256-byte
+// `SendMessages2` command header per append pass plus a 48-byte per-message
+// header (`server_common::send_messages2::{COMMAND_HEADER_SIZE,
+// MESSAGE_HEADER_SIZE}`), instead of the legacy 64-byte per-message header.
+// Each pass below sends all `MSGS_COUNT` messages in one batch.
+#[cfg(feature = "vsr")]
+const NG_BATCH_HEADER_SIZE: u64 = 256;
+#[cfg(feature = "vsr")]
+const NG_MESSAGE_HEADER_SIZE: u64 = 48;
+#[cfg(feature = "vsr")]
+const MSGS_SIZE: u64 =
+ NG_BATCH_HEADER_SIZE + (NG_MESSAGE_HEADER_SIZE + MESSAGE_PAYLOAD_SIZE_BYTES) * MSGS_COUNT;
pub async fn run(harness: &TestHarness) {
let client = harness
@@ -223,7 +247,17 @@ async fn validate_system_stats(
expected_size: u64,
expected_messages_count: u64,
) {
- let stats = client.get_stats().await.unwrap();
+ let deadline = Instant::now() + STATS_CONVERGENCE_TIMEOUT;
+ let stats = loop {
+ let stats = client.get_stats().await.unwrap();
+ if (stats.messages_count == expected_messages_count
+ && stats.messages_size_bytes.as_bytes_u64() == expected_size)
+ || Instant::now() >= deadline
+ {
+ break stats;
+ }
+ sleep(STATS_RETRY_INTERVAL).await;
+ };
assert_eq!(
stats.messages_count, expected_messages_count,
"system stats messages_count mismatch"
@@ -241,12 +275,22 @@ async fn validate_stream(
expected_size: u64,
expected_messages_count: u64,
) {
- // 1. Validate stream size and number of messages
- let stream = client
- .get_stream(&Identifier::from_str(stream_name).unwrap())
- .await
- .unwrap()
- .expect("Failed to get stream");
+ // 1. Fetch until the async commit-apply converges (see the retry note at
+ // the top of the file).
+ let deadline = Instant::now() + STATS_CONVERGENCE_TIMEOUT;
+ let stream = loop {
+ let stream = client
+ .get_stream(&Identifier::from_str(stream_name).unwrap())
+ .await
+ .unwrap()
+ .expect("Failed to get stream");
+ if (stream.size == expected_size && stream.messages_count == expected_messages_count)
+ || Instant::now() >= deadline
+ {
+ break stream;
+ }
+ sleep(STATS_RETRY_INTERVAL).await;
+ };
// 2. Validate stream size and number of messages
assert_eq!(stream.size, expected_size);
@@ -260,15 +304,25 @@ async fn validate_topic(
expected_size: u64,
expected_messages_count: u64,
) {
- // 1. Validate topic size and number of messages
- let topic = client
- .get_topic(
- &Identifier::from_str(stream_name).unwrap(),
- &Identifier::from_str(topic_name).unwrap(),
- )
- .await
- .unwrap()
- .expect("Failed to get topic");
+ // 1. Fetch until the async commit-apply converges (see the retry note at
+ // the top of the file).
+ let deadline = Instant::now() + STATS_CONVERGENCE_TIMEOUT;
+ let topic = loop {
+ let topic = client
+ .get_topic(
+ &Identifier::from_str(stream_name).unwrap(),
+ &Identifier::from_str(topic_name).unwrap(),
+ )
+ .await
+ .unwrap()
+ .expect("Failed to get topic");
+ if (topic.size == expected_size && topic.messages_count == expected_messages_count)
+ || Instant::now() >= deadline
+ {
+ break topic;
+ }
+ sleep(STATS_RETRY_INTERVAL).await;
+ };
// 2. Validate topic size and number of messages
assert_eq!(topic.size, expected_size);
diff --git a/core/integration/tests/server/scenarios/system_scenario.rs b/core/integration/tests/server/scenarios/system_scenario.rs
index f4839a603..c296ad631 100644
--- a/core/integration/tests/server/scenarios/system_scenario.rs
+++ b/core/integration/tests/server/scenarios/system_scenario.rs
@@ -23,6 +23,38 @@ use bytes::Bytes;
use iggy::prelude::*;
use integration::harness::{TestHarness, assert_clean_system};
use std::str::FromStr;
+use std::time::{Duration, Instant};
+use tokio::time::sleep;
+
+// The partition plane applies committed ops asynchronously on the owning
+// shard, so partition-level stats (segments/offset/messages/size) can lag a
+// commit by one apply pass. Fetch until `matches` holds (or the deadline
+// expires), then run the terminal assertions on the returned details.
+const TOPIC_CONVERGENCE_TIMEOUT: Duration = Duration::from_secs(10);
+const TOPIC_RETRY_INTERVAL: Duration = Duration::from_millis(100);
+
+async fn get_topic_when(
+ client: &IggyClient,
+ stream_name: &str,
+ topic_name: &str,
+ matches: impl Fn(&TopicDetails) -> bool,
+) -> TopicDetails {
+ let deadline = Instant::now() + TOPIC_CONVERGENCE_TIMEOUT;
+ loop {
+ let topic = client
+ .get_topic(
+ &Identifier::named(stream_name).unwrap(),
+ &Identifier::named(topic_name).unwrap(),
+ )
+ .await
+ .unwrap()
+ .expect("Failed to get topic");
+ if matches(&topic) || Instant::now() >= deadline {
+ return topic;
+ }
+ sleep(TOPIC_RETRY_INTERVAL).await;
+ }
+}
pub async fn run(harness: &TestHarness) {
let client = harness
@@ -120,15 +152,15 @@ pub async fn run(harness: &TestHarness) {
assert_eq!(topic.max_topic_size, MaxTopicSize::Unlimited);
assert_eq!(topic.replication_factor, 1);
- // 11. Get topic details by ID
- let topic = client
- .get_topic(
- &Identifier::named(STREAM_NAME).unwrap(),
- &Identifier::named(TOPIC_NAME).unwrap(),
- )
- .await
- .unwrap()
- .expect("Failed to get topic");
+ // 11. Get topic details by ID; the owning shards materialize the fresh
+ // partitions (first segment included) asynchronously after the commit.
+ let topic = get_topic_when(&client, STREAM_NAME, TOPIC_NAME, |topic| {
+ topic
+ .partitions
+ .iter()
+ .all(|partition| partition.segments_count == 1)
+ })
+ .await;
assert_eq!(topic.id, topic_id);
assert_eq!(topic.name, TOPIC_NAME);
assert_eq!(topic.partitions_count, PARTITIONS_COUNT);
@@ -244,20 +276,22 @@ pub async fn run(harness: &TestHarness) {
}
}
- // 20. Get topic details and validate the partition details
- let topic = client
- .get_topic(
- &Identifier::named(STREAM_NAME).unwrap(),
- &Identifier::named(TOPIC_NAME).unwrap(),
- )
- .await
- .unwrap()
- .expect("Failed to get topic");
+ // 20. Get topic details and validate the partition details, once the
+ // owning shard's commit-apply folds the send into the shared stats.
+ let topic = get_topic_when(&client, STREAM_NAME, TOPIC_NAME, |topic| {
+ topic.messages_count == u64::from(MESSAGES_COUNT)
+ })
+ .await;
assert_eq!(topic.id, topic_id);
assert_eq!(topic.name, TOPIC_NAME);
assert_eq!(topic.partitions_count, PARTITIONS_COUNT);
assert_eq!(topic.partitions.len(), PARTITIONS_COUNT as usize);
+ // The exact byte size is framing-specific: legacy counts a 64-byte header
+ // per message; server-ng counts its on-disk batch framing.
+ #[cfg(not(feature = "vsr"))]
assert_eq!(topic.size, 100502);
+ #[cfg(feature = "vsr")]
+ assert!(topic.size > 0);
assert_eq!(topic.messages_count, MESSAGES_COUNT as u64);
let topic_partition = topic.partitions.get((PARTITION_ID) as usize).unwrap();
assert_eq!(topic_partition.id, PARTITION_ID);
@@ -607,18 +641,29 @@ pub async fn run(harness: &TestHarness) {
.await
.unwrap();
- let polled_messages = client
- .poll_messages(
- &Identifier::named(STREAM_NAME).unwrap(),
- &Identifier::named(&updated_topic_name).unwrap(),
- Some(PARTITION_ID),
- &consumer,
- &PollingStrategy::offset(0),
- MESSAGES_COUNT,
- false,
- )
- .await
- .unwrap();
+ // The purge is applied by the owning shard's reconciler after the commit;
+ // poll until the wiped partition is observable.
+ let deadline = Instant::now() + TOPIC_CONVERGENCE_TIMEOUT;
+ let polled_messages = loop {
+ let polled_messages = client
+ .poll_messages(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(&updated_topic_name).unwrap(),
+ Some(PARTITION_ID),
+ &consumer,
+ &PollingStrategy::offset(0),
+ MESSAGES_COUNT,
+ false,
+ )
+ .await
+ .unwrap();
+ if (polled_messages.current_offset == 0 && polled_messages.messages.is_empty())
+ || Instant::now() >= deadline
+ {
+ break polled_messages;
+ }
+ sleep(TOPIC_RETRY_INTERVAL).await;
+ };
assert_eq!(polled_messages.current_offset, 0);
assert!(polled_messages.messages.is_empty());
@@ -658,18 +703,29 @@ pub async fn run(harness: &TestHarness) {
.await
.unwrap();
- let polled_messages = client
- .poll_messages(
- &Identifier::named(&updated_stream_name).unwrap(),
- &Identifier::named(&updated_topic_name).unwrap(),
- Some(PARTITION_ID),
- &consumer,
- &PollingStrategy::offset(0),
- MESSAGES_COUNT,
- false,
- )
- .await
- .unwrap();
+ // As with the topic purge above, the stream purge is applied by the
+ // owning shard's reconciler after the commit; poll until observable.
+ let deadline = Instant::now() + TOPIC_CONVERGENCE_TIMEOUT;
+ let polled_messages = loop {
+ let polled_messages = client
+ .poll_messages(
+ &Identifier::named(&updated_stream_name).unwrap(),
+ &Identifier::named(&updated_topic_name).unwrap(),
+ Some(PARTITION_ID),
+ &consumer,
+ &PollingStrategy::offset(0),
+ MESSAGES_COUNT,
+ false,
+ )
+ .await
+ .unwrap();
+ if (polled_messages.current_offset == 0 && polled_messages.messages.is_empty())
+ || Instant::now() >= deadline
+ {
+ break polled_messages;
+ }
+ sleep(TOPIC_RETRY_INTERVAL).await;
+ };
assert_eq!(polled_messages.current_offset, 0);
assert!(polled_messages.messages.is_empty());
@@ -725,15 +781,15 @@ pub async fn run(harness: &TestHarness) {
assert_eq!(topic.name, topic_name);
- // 45. Delete segments and validate partition segments count before and after
- let topic_before_delete = client
- .get_topic(
- &Identifier::named(&stream_name).unwrap(),
- &Identifier::named(&topic_name).unwrap(),
- )
- .await
- .unwrap()
- .expect("Failed to get topic");
+ // 45. Delete segments and validate partition segments count before and
+ // after; the recreated topic's partitions materialize asynchronously.
+ let topic_before_delete = get_topic_when(&client, &stream_name, &topic_name, |topic| {
+ topic
+ .partitions
+ .get(PARTITION_ID as usize)
+ .is_some_and(|partition| partition.segments_count > 0)
+ })
+ .await;
let segments_count_before_delete =
topic_before_delete.partitions[PARTITION_ID as usize].segments_count;
assert!(
diff --git a/core/integration/tests/server/specific.rs b/core/integration/tests/server/specific.rs
index dda7be085..baa8e4171 100644
--- a/core/integration/tests/server/specific.rs
+++ b/core/integration/tests/server/specific.rs
@@ -18,15 +18,11 @@
use crate::server::scenarios::{message_size_scenario, single_message_per_batch_scenario};
#[cfg(not(feature = "vsr"))]
use crate::server::scenarios::{reconnect_after_restart_scenario, restart_offset_skip_scenario};
-#[cfg(not(feature = "vsr"))]
use crate::server::scenarios::{
segment_rotation_race_scenario, tcp_tls_scenario, websocket_tls_scenario,
};
use integration::iggy_harness;
-// TLS listeners are not wired in server-ng's message bus (handshake fails
-// with BadSignature / eof and the SDK retries forever).
-#[cfg(not(feature = "vsr"))]
#[iggy_harness(
test_client_transport = TcpTlsGenerated,
server(tls = generated)
@@ -36,9 +32,6 @@ async fn tcp_tls_scenario_should_be_valid(harness: &TestHarness) {
tcp_tls_scenario::run(&client).await;
}
-// TLS listeners are not wired in server-ng's message bus (handshake fails
-// with BadSignature / eof and the SDK retries forever).
-#[cfg(not(feature = "vsr"))]
#[iggy_harness(
test_client_transport = TcpTlsSelfSigned,
server(tls = self_signed)
@@ -48,9 +41,6 @@ async fn tcp_tls_self_signed_scenario_should_be_valid(harness: &TestHarness) {
tcp_tls_scenario::run(&client).await;
}
-// TLS listeners are not wired in server-ng's message bus (handshake fails
-// with BadSignature / eof and the SDK retries forever).
-#[cfg(not(feature = "vsr"))]
#[iggy_harness(
test_client_transport = WebSocketTlsGenerated,
server(websocket_tls = generated)
@@ -149,10 +139,8 @@ async fn restart_offset_skip(harness: &mut TestHarness) {
/// Test configuration:
/// - 8 producers total (2 per protocol: TCP, HTTP, QUIC, WebSocket)
/// - All producers write to the same partition for maximum lock contention
-// Concurrency race test over all four transports (vsr has no HTTP) and
-// trips the metadata-plane consensus races; skip with the other
-// concurrent tests until those races are fixed.
-#[cfg(not(feature = "vsr"))]
+// Concurrency race test: under vsr it runs over the three VSR transports
+// (TCP/QUIC/WebSocket -- HTTP/REST carries no VSR framing), legacy runs all four.
#[iggy_harness(server(
segment.size = "512B",
message_saver.interval = "1s",
diff --git a/core/message_bus/Cargo.toml b/core/message_bus/Cargo.toml
index 4b63f2c93..a5017d127 100644
--- a/core/message_bus/Cargo.toml
+++ b/core/message_bus/Cargo.toml
@@ -32,6 +32,7 @@ publish = false
ahash = { workspace = true }
async-channel = { workspace = true }
blake3 = { workspace = true }
+bytemuck = { workspace = true }
bytes = { workspace = true }
compio = { workspace = true }
compio-quic = { workspace = true }
diff --git a/core/message_bus/src/client_listener/mod.rs b/core/message_bus/src/client_listener/mod.rs
index be3d9877a..2503be876 100644
--- a/core/message_bus/src/client_listener/mod.rs
+++ b/core/message_bus/src/client_listener/mod.rs
@@ -102,8 +102,10 @@ pub mod wss;
///
/// compio 0.19 replaced `TcpListener::bind_with_options(addr, SocketOpts)`
/// with the `TcpSocket` builder; this preserves the prior `nodelay(true)`
-/// bind. `SO_REUSEPORT` is intentionally not set: only shard 0 binds the
-/// client listeners (see each caller).
+/// bind. `SO_REUSEADDR` is set so a restarted server can rebind the port
+/// while a previous client connection lingers in `TIME_WAIT` (matching the
+/// replica and QUIC listeners); `SO_REUSEPORT` is intentionally not set:
+/// only shard 0 binds the client listeners (see each caller).
///
/// # Errors
///
@@ -120,6 +122,9 @@ pub async fn bind_nodelay_listener(
socket
.set_nodelay(true)
.map_err(|e| IggyError::IoError(e.to_string()))?;
+ socket
+ .set_reuseaddr(true)
+ .map_err(|e| IggyError::IoError(e.to_string()))?;
socket
.bind(addr)
.await
diff --git a/core/message_bus/src/transports/quic.rs b/core/message_bus/src/transports/quic.rs
index 02cb7dc7e..8fdf49e83 100644
--- a/core/message_bus/src/transports/quic.rs
+++ b/core/message_bus/src/transports/quic.rs
@@ -30,22 +30,31 @@
//! at a time:
//!
//! 1. accept the next bidi (or break on shutdown / connection close);
-//! 2. read exactly one inbound frame from the `RecvStream` and forward it
-//! to [`ActorContext::in_tx`];
-//! 3. await one outbound frame on [`ActorContext::rx`] (the reply), drain
-//! any additional immediate replies via `try_recv` (defensive — 1:1 in
-//! steady state), write each to the `SendStream`, then `finish()` to
-//! deliver pending data + FIN to the peer;
+//! 2. read exactly one inbound frame from the `RecvStream`, note its
+//! `request` id, and forward it to [`ActorContext::in_tx`];
+//! 3. wait on [`ActorContext::rx`] for the reply whose `request` id matches,
+//! discarding any stale/duplicate reply left by an earlier attempt's bidi
+//! (the SDK replays an explicit transient with the SAME request id); write
+//! the matched reply, then `finish()` to deliver pending data + FIN;
//! 4. drop the bidi pair and loop to the next `accept_bi`.
//!
-//! The `accept_bi` loop is serial by design: the single
-//! [`ActorContext::rx`] channel per connection is unkeyed, so two
-//! concurrent in-flight bidis would race the reply queue with no way to
-//! tell which reply belongs to which bidi. The iggy SDK on a single
-//! connection already serialises via an internal `RwLock`, so this is a
-//! semantic match, not a regression. A future pipelining upgrade would
-//! introduce `request_id`-keyed reply routing and run per-bidi handlers
-//! concurrently.
+//! The `accept_bi` loop is serial (the iggy SDK serialises per connection via
+//! an internal `RwLock`, so one bidi at a time is a semantic match) and the
+//! reply wait is `request_id`-KEYED. The wait is NOT bounded by a short
+//! abandon-timeout: every transient submit is answered with an explicit
+//! `TransientNotCommitted` rejection frame these days, so an unanswered bidi
+//! means the dispatch truly dropped the request (both partition queues full)
+//! or the op is pathologically slow -- and abandoning the bidi while the op
+//! can still commit would strand its reply. All partition ops on a
+//! connection share ONE request id (only metadata ops advance the dedup
+//! counter), so a stranded reply would be consumed by the NEXT partition
+//! op's bidi and shift the connection's data-plane stream off by one,
+//! permanently. Instead, [`REPLY_WAIT_BACKSTOP`] (longer than the SDK's
+//! whole-request deadline, so the client always gives up first and
+//! reconnects) closes the CONNECTION, which drops the mailbox and every
+//! pending reply with it -- nothing can strand or cross request ids. A
+//! future pipelining upgrade would additionally run per-bidi handlers
+//! concurrently (and would then require a real per-op correlation id).
//!
//! The handshake itself is driven by [`accept_handshake`] (no bidi
//! captured). To avoid a single slow / hostile peer wedging the entire
@@ -73,6 +82,7 @@
use super::{ActorContext, TransportConn, TransportListener};
use crate::config::QuicTuning;
use crate::framing;
+use crate::lifecycle::connection_registry::BusMessage;
use compio::BufResult;
use compio::io::AsyncWriteExt;
use compio_quic::{
@@ -80,6 +90,7 @@ use compio_quic::{
crypto::rustls::QuicServerConfig,
};
use futures::FutureExt;
+use iggy_binary_protocol::{Command2, HEADER_SIZE, ReplyHeader, RequestHeader};
use std::io;
use std::net::SocketAddr;
use std::sync::Arc;
@@ -276,6 +287,54 @@ async fn accept_handshake_inner(incoming: Incoming) -> Option<(QuicTransportConn
/// using `MessageBusConfig::close_grace`.
pub(crate) const DEFAULT_CLOSE_GRACE: Duration = Duration::from_secs(2);
+/// How long a bidi waits for its keyed reply before abandoning the stream.
+/// A replicated submit the server cannot yet commit is answered with silence
+/// (the server expects the SDK to replay), so without a bound the per-bidi
+/// wait - and, since `accept_bi` is serial, the whole connection - would wedge:
+/// Backstop for a bidi whose dispatch never produced a reply (a true drop:
+/// both partition queues full, or a pathologically slow commit). Longer than
+/// the SDK's whole-request deadline (30s), so the client always times out
+/// first, surfaces `Disconnected`, and reconnects. On firing the CONNECTION
+/// is closed rather than the bidi abandoned: partition ops all share one
+/// request id, so a reply stranded by an abandoned bidi would be consumed by
+/// the next data-plane op and shift every subsequent reply off by one.
+/// Closing the connection drops the mailbox and any late reply with it.
+const REPLY_WAIT_BACKSTOP: Duration = Duration::from_mins(1);
+
+/// `Command2` byte offset shared by every consensus header (after the
+/// fixed checksum/cluster/size/view/release prefix).
+const COMMAND_OFFSET: usize = 60;
+
+/// How an outbound frame routes to a waiting bidi.
+enum ReplyRoute {
+ /// A request-keyed `Reply`: deliver only to the bidi whose request id
+ /// matches; a non-matching one is a stale/duplicate from an abandoned bidi.
+ Keyed(u64),
+ /// A terminal/control frame (`Eviction`, etc.) with no `request` field, or
+ /// an undecodable header: deliver to the in-flight bidi unconditionally.
+ /// Filtering these out would silently swallow session-terminal signals
+ /// (e.g. `NoSession` -> `Unauthenticated`), leaving the client to time out.
+ Unkeyed,
+}
+
+/// Decide how to route an outbound frame. Only `Command2::Reply` carries a
+/// `request` id; everything else must pass through unfiltered.
+fn reply_route(frame: &BusMessage) -> ReplyRoute {
+ let bytes = frame.as_slice();
+ let is_reply = bytes
+ .get(COMMAND_OFFSET)
+ .is_some_and(|&command| command == Command2::Reply as u8);
+ if !is_reply {
+ return ReplyRoute::Unkeyed;
+ }
+ bytes
+ .get(..HEADER_SIZE)
+ .and_then(|header| bytemuck::checked::try_from_bytes::<ReplyHeader>(header).ok())
+ .map_or(ReplyRoute::Unkeyed, |header| {
+ ReplyRoute::Keyed(header.request)
+ })
+}
+
/// A single accepted QUIC connection.
///
/// Per-request bidirectional streams are accepted on demand inside
@@ -317,7 +376,7 @@ impl QuicTransportConn {
}
impl TransportConn for QuicTransportConn {
- #[allow(clippy::future_not_send)]
+ #[allow(clippy::future_not_send, clippy::too_many_lines)]
async fn run(self, ctx: ActorContext) {
let connection = self.connection;
let ActorContext {
@@ -369,46 +428,89 @@ impl TransportConn for QuicTransportConn {
continue;
}
};
+ // Request id this bidi is waiting on. Used to match its reply and
+ // to discard stale/duplicate replies left by an abandoned bidi.
+ // `None` (unparsable) degrades to accepting the first reply.
+ let request_id = req
+ .as_slice()
+ .get(..HEADER_SIZE)
+ .and_then(|bytes| bytemuck::checked::try_from_bytes::<RequestHeader>(bytes).ok())
+ .map(|header| header.request);
+
if in_tx.send(req).await.is_err() {
debug!(%label, %peer, "quic: inbound queue dropped");
break;
}
- // Wait for the matching reply. The single bus rx channel is
- // unkeyed; serialising one bidi at a time guarantees the
- // next frame on `rx` is the reply for THIS bidi.
- // `async_channel::Receiver::recv` and `FusedShutdown` are
- // both cancel-safe and compio-buffer-free, so select-drop
- // here cannot poison anything.
- let first = futures::select! {
- () = shutdown_fut.as_mut() => {
- debug!(%label, %peer, "quic: shutdown during reply wait");
- break;
+ // Wait for THIS bidi's reply, keyed by request id. The bus `rx` is
+ // a single unkeyed per-connection channel, so filter by id and
+ // discard replies belonging to an earlier attempt's bidi (an
+ // explicit-transient replay reuses the SAME request id, so a late
+ // reply matches and is consumed by the replay's bidi - never
+ // misrouted to a later request). There is no short abandon-timeout:
+ // transient submits are answered with explicit rejection frames, so
+ // silence here means the dispatch dropped the request or the op is
+ // pathologically slow. `REPLY_WAIT_BACKSTOP` (see its doc) closes
+ // the whole connection instead of abandoning the bidi, so a late
+ // reply can never strand and shift onto the next data-plane op.
+ let mut matched: Option<BusMessage> = None;
+ let mut connection_gone = false;
+ {
+ let mut reply_deadline =
+ std::pin::pin!(compio::time::sleep(REPLY_WAIT_BACKSTOP).fuse());
+ loop {
+ futures::select! {
+ () = shutdown_fut.as_mut() => {
+ debug!(%label, %peer, "quic: shutdown during reply wait");
+ connection_gone = true;
+ break;
+ }
+ () = reply_deadline.as_mut() => {
+ warn!(%label, %peer, ?request_id, "quic: no reply within backstop; closing connection so no reply can strand");
+ connection_gone = true;
+ break;
+ }
+ res = rx.recv().fuse() => if let Ok(m) = res {
+ let deliver = match reply_route(&m) {
+ // Eviction / control frame: always deliver.
+ ReplyRoute::Unkeyed => true,
+ // Keyed reply: only this bidi's; else it's a
+ // stale/duplicate from an abandoned bidi -> drop.
+ ReplyRoute::Keyed(id) => {
+ request_id.is_none() || Some(id) == request_id
+ }
+ };
+ if deliver {
+ matched = Some(m);
+ break;
+ }
+ } else {
+ debug!(%label, %peer, "quic: mailbox closed");
+ connection_gone = true;
+ break;
+ },
+ }
}
- res = rx.recv().fuse() => if let Ok(m) = res {
- m
- } else {
- debug!(%label, %peer, "quic: mailbox closed");
- break;
- },
+ }
+ if connection_gone {
+ break;
+ }
+ let Some(first) = matched else {
+ // Unreachable in practice: every no-reply path above sets
+ // `connection_gone`. Close the send half defensively.
+ if let Err(e) = send.finish() {
+ debug!(%label, %peer, error = ?e, "quic: send.finish() failed with no reply");
+ }
+ continue;
};
- // Write the reply, drain any additional immediate replies
- // (defensive 1:N), then `finish()` so quinn-proto flushes
- // pending data + a FIN, signalling the SDK that the reply
- // is complete.
+ // Write the reply, then `finish()` so quinn-proto flushes pending
+ // data + a FIN, signalling the SDK that the reply is complete.
let BufResult(result, _frozen) = send.write_all(first).await;
if let Err(e) = result {
debug!(%label, %peer, error = ?e, "quic: write failed");
continue;
}
- while let Ok(more) = rx.try_recv() {
- let BufResult(result, _frozen) = send.write_all(more).await;
- if let Err(e) = result {
- debug!(%label, %peer, error = ?e, "quic: write failed during burst drain");
- break;
- }
- }
if let Err(e) = send.finish() {
debug!(%label, %peer, error = ?e, "quic: send.finish() failed");
}
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index 73511c701..b331cafc9 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -26,8 +26,8 @@ use consensus::{
PipelineEntry, Plane, PlaneIdentity, PlaneKind, PreflightOutcome, Project, ReplicaLogContext,
RequestLogEvent, Sequencer, SimEventKind, VsrConsensus, ack_preflight, ack_quorum_reached,
apply_preflight_consensus_plane, build_eviction_message, build_reply_message,
- build_reply_message_with, drain_committable_prefix, emit_sim_event,
- fence_old_prepare_by_commit, is_caught_up_primary,
+ build_reply_message_with, build_result_rejection_reply, drain_committable_prefix,
+ emit_sim_event, fence_old_prepare_by_commit, is_caught_up_primary,
panic_if_hash_chain_would_break_in_same_view, pipeline_prepare_common, register_preflight,
replicate_preflight, replicate_to_next_in_chain, request_preflight, send_eviction_to_client,
send_prepare_ok as send_prepare_ok_common,
@@ -38,6 +38,7 @@ use iggy_binary_protocol::requests::partitions::CreatePartitionsRequest as WireC
use iggy_binary_protocol::requests::partitions::CreatePartitionsWithAssignmentsRequest as PersistedCreatePartitionsRequest;
use iggy_binary_protocol::requests::topics::CreateTopicRequest as WireCreateTopicRequest;
use iggy_binary_protocol::requests::topics::CreateTopicWithAssignmentsRequest as PersistedCreateTopicRequest;
+use iggy_binary_protocol::requests::topics::UpdateTopicRequest as WireUpdateTopicRequest;
use iggy_binary_protocol::{
Command2, ConsensusHeader, EvictionReason, GenericHeader, Operation, PrepareHeader,
PrepareOkHeader, ReplyHeader, RequestHeader, WireDecode, WireEncode, WireName,
@@ -48,7 +49,7 @@ use iggy_common::variadic;
use journal::{Journal, JournalHandle};
use message_bus::MessageBus;
use server_common::Message;
-use std::cell::RefCell;
+use std::cell::{Cell, RefCell};
use std::mem::size_of;
use std::path::Path;
use std::rc::Rc;
@@ -381,6 +382,13 @@ pub struct IggyMetadata<C, J, S, M> {
/// [`Self::set_commit_notifier`] runs (server-ng bootstrap on shard
/// 0 sets it; peer shards and tests leave it `None`).
commit_notifier: RefCell<Option<CommitNotifier>>,
+ /// Resolved byte value for `MaxTopicSize::ServerDefault` (`0` on the
+ /// wire). Primary admission rewrites the sentinel to this value before
+ /// replication so the committed state carries a concrete size and every
+ /// replica resolves identically regardless of local config. Set from
+ /// server config at bootstrap ([`Self::set_default_max_topic_size`]);
+ /// defaults to unlimited, matching the shipped server config.
+ default_max_topic_size: Cell<u64>,
}
impl<C, J, S, M> IggyMetadata<C, J, S, M>
@@ -411,6 +419,7 @@ where
coordinator,
client_table: RefCell::new(ClientTable::new(CLIENTS_TABLE_MAX)),
commit_notifier: RefCell::new(None),
+ default_max_topic_size: Cell::new(u64::MAX),
}
}
}
@@ -423,6 +432,19 @@ impl<C, J, S, M> IggyMetadata<C, J, S, M> {
*self.commit_notifier.borrow_mut() = notifier;
}
+ /// Install the resolved byte value used for `MaxTopicSize::ServerDefault`.
+ /// Server-ng bootstrap calls this with `system.topic.max_size` on every
+ /// shard (responses read it too); only shard 0's copy feeds admission.
+ pub fn set_default_max_topic_size(&self, max_topic_size_bytes: u64) {
+ self.default_max_topic_size.set(max_topic_size_bytes);
+ }
+
+ /// Resolved byte value for `MaxTopicSize::ServerDefault`.
+ #[must_use]
+ pub const fn default_max_topic_size(&self) -> u64 {
+ self.default_max_topic_size.get()
+ }
+
/// Fire post-commit notifier. Clones the `Rc` out under a short
/// borrow so a re-entrant `set_commit_notifier` from inside the
/// closure cannot panic on `borrow_mut`.
@@ -530,6 +552,7 @@ where
.await;
}
+ #[allow(clippy::too_many_lines)]
async fn on_replicate(&self, message: <VsrConsensus<B> as Consensus>::Message<PrepareHeader>) {
let Some(consensus) =
require_shard_zero(self.consensus.as_ref(), "on_replicate", "consensus")
@@ -560,11 +583,37 @@ where
}
};
- // TODO: Handle idx calculation, for now using header.op, but since the journal may get compacted, this may not be correct.
+ // Fenced by commit: the whole chain has already committed this op, so
+ // nobody needs it again. Drop entirely. (Mirror of the partition
+ // plane's split in `IggyPartition::on_replicate`.)
+ #[allow(clippy::cast_possible_truncation)]
+ if fence_old_prepare_by_commit(consensus, &header) {
+ warn!(
+ target: "iggy.metadata.diag",
+ plane = "metadata",
+ replica_id = consensus.replica(),
+ view = consensus.view(),
+ op = header.op,
+ commit = consensus.commit_max(),
+ operation = ?header.operation,
+ "received old prepare (<= commit), skipping replication"
+ );
+ return;
+ }
+
+ // Durable here but not yet committed, and the primary is retransmitting
+ // it: our original PrepareOk was lost (e.g. the primary's inbox
+ // overflowed under a client burst). Re-forward the tail down the chain
+ // so a downstream replica that missed it recovers, then re-ack ONLY
+ // the retransmitted op. The primary's retransmit cycle walks every
+ // un-acked op in the window (`retransmit_targets`), so a lost ack for
+ // a lower op gets its own retransmit and its own re-ack; re-acking the
+ // whole suffix here is O(window^2) PrepareOks per cycle across the
+ // backups, which can overflow the primary's inbox -- the very failure
+ // this path recovers from. Both downstream and primary are idempotent
+ // on a duplicate (replica, op).
#[allow(clippy::cast_possible_truncation)]
- let is_old_prepare = fence_old_prepare_by_commit(consensus, &header)
- || journal.handle().header(header.op as usize).is_some();
- if is_old_prepare {
+ if journal.handle().header(header.op as usize).is_some() {
warn!(
target: "iggy.metadata.diag",
plane = "metadata",
@@ -573,10 +622,10 @@ where
op = header.op,
commit = consensus.commit_max(),
operation = ?header.operation,
- "received old prepare, skipping replication"
+ "journal already holds prepare, re-forwarding + re-acking it"
);
- // Old prepare: downstream already has it or learns via newer
- // forward; no chain-replicate; WAL unaffected.
+ self.replicate(&message).await;
+ self.send_prepare_ok(&header).await;
return;
}
@@ -1303,22 +1352,27 @@ where
.as_ref()
.expect("submit_request_in_process: consensus only exists on shard 0");
+ // Not-primary / not-caught-up is transient: the same request replayed
+ // once a primary is caught up commits fine. Reply with the explicit
+ // `TransientNotCommitted` frame (relayed to the socket by the home shard)
+ // so the client replays immediately rather than waiting out its
+ // read-timeout. The frame keeps the lockstep TCP/WS stream in framing
+ // sync, so the client resends on the same connection (session intact).
if !is_caught_up_primary(consensus) {
- return Err(
- if consensus.is_primary() && consensus.is_normal() && !consensus.is_syncing() {
- MetadataSubmitError::NotCaughtUp
- } else {
- MetadataSubmitError::NotPrimary
- },
- );
+ return Ok(build_result_rejection_reply(
+ &request_header,
+ consensus.commit_max(),
+ IggyError::TransientNotCommitted.as_code(),
+ )
+ .into_generic());
}
// Dedup / session / eviction. shard 0 cannot route by the VSR
// consensus `client_id` (its top bits are random, not home-shard
- // routing), so a Replay/Evict is returned to the home shard as the
- // reply -- `handle_client_request` writes it to the originating socket
- // by transport id, exactly like a fresh commit. Drop surfaces as
- // Canceled so the home shard stays silent and the SDK replays.
+ // routing), so a Replay/Evict/NotReady is returned to the home shard as
+ // the reply -- `handle_client_request` writes it to the originating
+ // socket by transport id, exactly like a fresh commit. Drop (client-bug
+ // stale/gap) surfaces as Canceled so the home shard stays silent.
match request_preflight(consensus, &self.client_table, client_id, session, request) {
PreflightOutcome::Dispatch => {}
PreflightOutcome::Replay(reply) => {
@@ -1333,11 +1387,27 @@ where
let ctx = EvictionContext::from_consensus(consensus);
return Ok(build_eviction_message(ctx, client_id, reason).into_generic());
}
+ // In-flight prepare from this client: replaying the same request_id
+ // is absorbed until the original commits, then served from cache.
+ PreflightOutcome::NotReady => {
+ return Ok(build_result_rejection_reply(
+ &request_header,
+ consensus.commit_max(),
+ IggyError::TransientNotCommitted.as_code(),
+ )
+ .into_generic());
+ }
PreflightOutcome::Drop => return Err(MetadataSubmitError::Canceled),
}
+ // Pipeline full: backpressure, not failure. Tell the client to replay.
if consensus.pipeline().borrow().is_full() {
- return Err(MetadataSubmitError::PipelineFull);
+ return Ok(build_result_rejection_reply(
+ &request_header,
+ consensus.commit_max(),
+ IggyError::TransientNotCommitted.as_code(),
+ )
+ .into_generic());
}
let Ok(prepare) = self.prepare_request(message) else {
@@ -1349,10 +1419,19 @@ where
return Ok(build_eviction_message(ctx, client_id, reason).into_generic());
};
- self.dispatch_prepare_and_await(consensus, prepare)
- .await
- .map(server_common::Message::into_generic)
- .map_err(|Canceled| MetadataSubmitError::Canceled)
+ // A view change canceled the pending prepare before commit. The op may
+ // or may not have committed; replaying the same request_id is idempotent
+ // (the new primary serves it from cache if committed, else re-dispatches),
+ // so reply with the transient frame rather than staying silent.
+ match self.dispatch_prepare_and_await(consensus, prepare).await {
+ Ok(reply) => Ok(reply.into_generic()),
+ Err(Canceled) => Ok(build_result_rejection_reply(
+ &request_header,
+ consensus.commit_max(),
+ IggyError::TransientNotCommitted.as_code(),
+ )
+ .into_generic()),
+ }
}
/// Subscribe to a prepared metadata write, dispatch it into the pipeline,
@@ -1410,6 +1489,112 @@ where
receiver.await
}
+ /// Repair the primary's own missing self-acks.
+ ///
+ /// The primary's `PrepareOk` for its own prepare is produced exactly once,
+ /// as a loopback right after the WAL append (see `on_replicate`). If that
+ /// one-shot is lost or suppressed (e.g. the `send_prepare_ok` persistence
+ /// gate races the sequencer pre-advance under a client burst), no
+ /// retransmit path regenerates it: `retransmit_targets` lists the primary
+ /// itself among the missing replicas, but `RetransmitPrepares` to self is a
+ /// no-op. The op then sits one vote short of quorum forever and pins the
+ /// contiguous commit prefix, so `commit_min` never catches up to
+ /// `commit_max` and the cluster wedges.
+ ///
+ /// This is a re-ack-only repair: for each pending op the primary holds
+ /// DURABLY but has not self-acked, re-emit the self `PrepareOk` and drain
+ /// it through `on_ack`. A pending op the primary does NOT yet hold durably
+ /// is skipped - filling that hole needs full message repair, which is out
+ /// of scope here. Driven each consensus tick;
+ /// `on_ack` dedups a redundant self-ack via `has_ack`, so re-running is
+ /// idempotent and stops once the op commits and leaves the pending range.
+ #[allow(clippy::future_not_send, clippy::cast_possible_truncation)]
+ pub async fn repair_primary_self_acks(&self) {
+ let Some(consensus) = self.consensus.as_ref() else {
+ return;
+ };
+ if !consensus.is_primary() || !consensus.is_normal() || consensus.is_syncing() {
+ return;
+ }
+ let Some(journal) = self.journal.as_ref() else {
+ return;
+ };
+ let self_replica = consensus.replica();
+
+ // Snapshot durable, self-unacked pending ops, dropping the pipeline and
+ // journal borrows before the `send_prepare_ok` awaits below.
+ let mut headers: Vec<PrepareHeader> = Vec::new();
+ {
+ let pipeline = consensus.pipeline().borrow();
+ let from = consensus.commit_max() + 1;
+ let to = consensus.sequencer().current_sequence();
+ for op in from..=to {
+ let Some(entry) = pipeline.entry_by_op(op) else {
+ continue;
+ };
+ if entry.has_ack(self_replica) {
+ continue;
+ }
+ // Durable only: re-acking implies "I hold this op". A gap (op not
+ // in the journal) must not be self-acked - that path needs repair.
+ if let Some(header) = journal.handle().header(op as usize).map(|header| *header) {
+ headers.push(header);
+ }
+ }
+ }
+ if headers.is_empty() {
+ return;
+ }
+
+ // Interleave push + drain per header instead of push-all-then-drain-once:
+ // each `on_ack` below can promote a full window of buffered requests
+ // (`drain_request_queue_into_prepares`), and every promoted prepare
+ // self-acks through `send_or_loopback(self)` -> `push_loopback`. The
+ // consensus-tick arm of the shard pump never drains the loopback
+ // queue, so residuals would accumulate across ticks and trip the
+ // `push_loopback` capacity assert (`PIPELINE_PREPARE_QUEUE_MAX`).
+ // Draining to empty BEFORE each push bounds queue occupancy to one
+ // promotion window; the trailing drain applies the acks this pass
+ // produced (including promotion self-acks) instead of leaving them
+ // for a tick that never comes.
+ let mut loopback = Vec::new();
+ for header in &headers {
+ while self.apply_self_ack_loopback(&mut loopback).await {}
+ self.send_prepare_ok(header).await;
+ }
+ while self.apply_self_ack_loopback(&mut loopback).await {}
+ }
+
+ /// Drain the consensus loopback queue once and feed every self-`PrepareOk`
+ /// through [`Self::on_ack`], dropping anything else with a warning.
+ /// Returns whether any message was processed, so callers can loop until
+ /// the queue is empty (an `on_ack` can promote buffered requests whose
+ /// self-acks land back on the queue).
+ #[allow(clippy::future_not_send)]
+ async fn apply_self_ack_loopback(&self, loopback: &mut Vec<Message<GenericHeader>>) -> bool {
+ let consensus = self.consensus.as_ref().unwrap();
+ consensus.drain_loopback_into(loopback);
+ if loopback.is_empty() {
+ return false;
+ }
+ for message in loopback.drain(..) {
+ match message.header().command {
+ Command2::PrepareOk => match message.try_into_typed::<PrepareOkHeader>() {
+ Ok(prepare_ok) => self.on_ack(prepare_ok).await,
+ Err(error) => warn!(
+ error = %error,
+ "dropping malformed PrepareOk from self-ack repair loopback"
+ ),
+ },
+ command => warn!(
+ ?command,
+ "dropping unexpected message from self-ack repair loopback"
+ ),
+ }
+ }
+ true
+ }
+
/// Promote up to `slots_freed` buffered requests into prepares after
/// `on_ack` commits a prefix.
///
@@ -1541,8 +1726,14 @@ where
match header.operation {
Operation::CreateTopic => {
- let request = WireCreateTopicRequest::decode_from(body)
+ let mut request = WireCreateTopicRequest::decode_from(body)
.map_err(|_| IggyError::InvalidCommand)?;
+ // Resolve the `ServerDefault` sentinel (0) against server config
+ // here, at primary admission, so the replicated payload carries a
+ // concrete size and every replica commits the same value.
+ if request.max_topic_size == 0 {
+ request.max_topic_size = self.default_max_topic_size.get();
+ }
let partitions = self
.allocator
.allocate_many(request.partitions_count as usize)
@@ -1601,6 +1792,22 @@ where
&body,
))
}
+ Operation::UpdateTopic => {
+ let mut request = WireUpdateTopicRequest::decode_from(body)
+ .map_err(|_| IggyError::InvalidCommand)?;
+ if request.max_topic_size == 0 {
+ // Same `ServerDefault` resolution as `CreateTopic` above.
+ request.max_topic_size = self.default_max_topic_size.get();
+ let body = request.to_bytes();
+ return Ok(build_prepare_message(
+ consensus,
+ &header,
+ Operation::UpdateTopic,
+ &body,
+ ));
+ }
+ Ok(message.project(consensus))
+ }
_ => Ok(message.project(consensus)),
}
}
diff --git a/core/metadata/src/stm/consumer_group.rs b/core/metadata/src/stm/consumer_group.rs
index b05afb873..e99b16fdb 100644
--- a/core/metadata/src/stm/consumer_group.rs
+++ b/core/metadata/src/stm/consumer_group.rs
@@ -523,12 +523,13 @@ impl StateHandler for CreateConsumerGroupRequest {
.consumer_groups
.insert(id, ConsumerGroup::new(id, name.clone()));
topic.consumer_group_index.insert(name, id);
+ let partitions_count = topic.partitions.len() as u32;
ApplyReply::ok(
ConsumerGroupDetailsResponse {
group: ConsumerGroupResponse {
id: id as u32,
- partitions_count: 0,
+ partitions_count,
members_count: 0,
name: self.name.clone(),
},
diff --git a/core/metadata/src/stm/stream.rs b/core/metadata/src/stm/stream.rs
index 765e643df..3844706be 100644
--- a/core/metadata/src/stm/stream.rs
+++ b/core/metadata/src/stm/stream.rs
@@ -52,7 +52,8 @@ use iggy_binary_protocol::responses::streams::get_stream::{GetStreamResponse, To
use iggy_binary_protocol::responses::topics::get_topic::PartitionResponse;
use iggy_binary_protocol::{WireIdentifier, WireName};
use iggy_common::{
- CompressionAlgorithm, IggyExpiry, IggyTimestamp, MaxTopicSize, StreamStats, TopicStats,
+ CompressionAlgorithm, IggyExpiry, IggyTimestamp, MaxTopicSize, PartitionStats, StreamStats,
+ TopicStats,
};
use serde::{Deserialize, Serialize};
use server_common::sharding::IggyNamespace;
@@ -329,6 +330,131 @@ impl Stream {
}
}
+/// Cross-buffer shared aggregate stats.
+///
+/// The metadata STM is left-right double-buffered; `create_stream` /
+/// `create_topic` / snapshot restore run on BOTH buffers (Absorb
+/// re-dispatches each op), which would mint a distinct `Arc<StreamStats>` /
+/// `Arc<TopicStats>` per buffer. Aggregate message/size counters are
+/// `AtomicU64`s the partition plane increments DIRECTLY (outside the consensus
+/// op-log), so per-buffer Arcs silently drop those increments on a buffer swap
+/// (a read lands on the other, un-incremented buffer -- hence the
+/// `messages_count_inconsistent` naming). This registry hands both buffers the
+/// SAME `Arc` (get-or-create by id), so a direct increment is visible on every
+/// read.
+///
+/// Shared across buffers and reader shards via `Arc` (a `StreamsInner` clone
+/// shares it). Only shard 0's writer mutates the maps, under the
+/// single-threaded Absorb; the `Mutex` is for `Sync` (uncontended), not
+/// concurrency. Ids are deterministic across replicas (same op order), so both
+/// buffers resolve the same key.
+#[derive(Debug, Default)]
+pub struct StatsRegistry {
+ streams: std::sync::Mutex<AHashMap<usize, Arc<StreamStats>>>,
+ topics: std::sync::Mutex<AHashMap<(usize, usize), Arc<TopicStats>>>,
+ partitions: std::sync::Mutex<AHashMap<(usize, usize, usize), Arc<PartitionStats>>>,
+}
+
+impl StatsRegistry {
+ fn stream(&self, id: usize) -> Arc<StreamStats> {
+ self.streams
+ .lock()
+ .expect("stats registry mutex poisoned")
+ .entry(id)
+ .or_insert_with(|| Arc::new(StreamStats::default()))
+ .clone()
+ }
+
+ fn topic(
+ &self,
+ stream_id: usize,
+ topic_id: usize,
+ parent: Arc<StreamStats>,
+ ) -> Arc<TopicStats> {
+ self.topics
+ .lock()
+ .expect("stats registry mutex poisoned")
+ .entry((stream_id, topic_id))
+ .or_insert_with(|| Arc::new(TopicStats::new(parent)))
+ .clone()
+ }
+
+ /// Get-or-create the shared per-partition stats. The owning shard calls
+ /// this when it materializes the partition; any shard's `get_topic` reply
+ /// reads the same `Arc`, so partition-plane counters are visible
+ /// cross-shard without a gather.
+ ///
+ /// # Panics
+ /// If the registry mutex is poisoned.
+ pub fn partition(
+ &self,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+ parent: Arc<TopicStats>,
+ ) -> Arc<PartitionStats> {
+ self.partitions
+ .lock()
+ .expect("stats registry mutex poisoned")
+ .entry((stream_id, topic_id, partition_id))
+ .or_insert_with(|| Arc::new(PartitionStats::new(parent)))
+ .clone()
+ }
+
+ /// Read-only lookup for reply builders: `None` until the owning shard
+ /// materializes the partition.
+ ///
+ /// # Panics
+ /// If the registry mutex is poisoned.
+ pub fn partition_get(
+ &self,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+ ) -> Option<Arc<PartitionStats>> {
+ self.partitions
+ .lock()
+ .expect("stats registry mutex poisoned")
+ .get(&(stream_id, topic_id, partition_id))
+ .cloned()
+ }
+
+ fn remove_stream(&self, id: usize) {
+ self.streams
+ .lock()
+ .expect("stats registry mutex poisoned")
+ .remove(&id);
+ self.topics
+ .lock()
+ .expect("stats registry mutex poisoned")
+ .retain(|(stream_id, _), _| *stream_id != id);
+ self.partitions
+ .lock()
+ .expect("stats registry mutex poisoned")
+ .retain(|(stream_id, _, _), _| *stream_id != id);
+ }
+
+ fn remove_topic(&self, stream_id: usize, topic_id: usize) {
+ self.topics
+ .lock()
+ .expect("stats registry mutex poisoned")
+ .remove(&(stream_id, topic_id));
+ self.partitions
+ .lock()
+ .expect("stats registry mutex poisoned")
+ .retain(|(sid, tid, _), _| !(*sid == stream_id && *tid == topic_id));
+ }
+
+ fn remove_partitions_from(&self, stream_id: usize, topic_id: usize, first_removed: usize) {
+ self.partitions
+ .lock()
+ .expect("stats registry mutex poisoned")
+ .retain(|(sid, tid, pid), _| {
+ !(*sid == stream_id && *tid == topic_id && *pid >= first_removed)
+ });
+ }
+}
+
define_state! {
Streams {
index: AHashMap<Arc<str>, usize>,
@@ -345,6 +471,10 @@ define_state! {
// decide whether to wake the reconciler. Deterministic (same ops, same
// recompute on every replica).
pending_revocations_count: u64,
+ // Shared aggregate stats, one `Arc` per stream/topic across both
+ // left-right buffers (see `StatsRegistry`). Not snapshotted -- rebuilt
+ // as streams/topics restore.
+ stats_registry: Arc<StatsRegistry>,
}
}
@@ -1046,19 +1176,21 @@ impl StateHandler for CreateStreamRequest {
return ApplyReply::err(CreateStreamResult::NameAlreadyExists);
}
+ // Share one `Arc<StreamStats>` across both left-right buffers via the
+ // registry (see `StatsRegistry`). The id the next insert will use is
+ // deterministic, so both buffers resolve the same registry key.
+ let id = state.items.vacant_key();
+ let stream_stats = state.stats_registry.stream(id);
let stream = Stream {
- id: 0,
+ id,
name: name_arc.clone(),
created_at: timestamp,
- stats: Arc::new(StreamStats::default()),
+ stats: stream_stats,
topics: Slab::new(),
topic_index: AHashMap::default(),
};
-
- let id = state.items.insert(stream);
- if let Some(stream) = state.items.get_mut(id) {
- stream.id = id;
- }
+ let inserted = state.items.insert(stream);
+ debug_assert_eq!(inserted, id, "vacant_key must match the insert slot");
state.index.insert(name_arc, id);
// Reply body: a freshly created stream has no topics. The SDK
@@ -1118,6 +1250,8 @@ impl StateHandler for DeleteStreamRequest {
let name = stream.name.clone();
state.items.remove(stream_id);
state.index.remove(&name);
+ // Evict registry entries so a reused slab id starts with fresh stats.
+ state.stats_registry.remove_stream(stream_id);
state.revision = state.revision.wrapping_add(1);
// The dropped stream may have held groups with pending revocations.
state.recompute_pending_revocations_count();
@@ -1128,15 +1262,29 @@ impl StateHandler for DeleteStreamRequest {
impl StateHandler for PurgeStreamRequest {
type State = StreamsInner;
fn apply(&self, state: &mut StreamsInner, _timestamp: IggyTimestamp) -> ApplyReply {
- // Message data lives in the partition plane, not metadata. A purge leaves
- // the metadata shape (streams, topics, partition ids) intact, so apply
- // only validates the parent and acks: no mutation, no `revision` bump
- // (the reconciler keys off partition shape, which a purge preserves). The
- // partition-journal segment drop is not yet wired off this committed op,
- // so a committed purge is a metadata-plane no-op for now.
- // TODO: partition-plane purge off the committed op.
- if state.resolve_stream_id(&self.stream_id).is_none() {
- return ApplyReply::err(PurgeStreamResult::StreamNotFound);
+ // Stream purge = topic purge over every topic in the stream: advance
+ // each partition's monotonic purge generation; every replica's
+ // reconciler observes the committed generation and resets the
+ // partition to a single empty segment at offset 0 with cleared
+ // offsets (see `PurgeTopicRequest`). Metadata shape stays intact.
+ let advanced = {
+ let Some(stream_id) = state.resolve_stream_id(&self.stream_id) else {
+ return ApplyReply::err(PurgeStreamResult::StreamNotFound);
+ };
+ let Some(stream) = state.items.get_mut(stream_id) else {
+ return ApplyReply::err(PurgeStreamResult::StreamNotFound);
+ };
+ let mut advanced = false;
+ for (_, topic) in &mut stream.topics {
+ for partition in &mut topic.partitions {
+ partition.purge_generation = partition.purge_generation.wrapping_add(1);
+ advanced = true;
+ }
+ }
+ advanced
+ };
+ if advanced {
+ state.revision = state.revision.wrapping_add(1);
}
ApplyReply::ok(Bytes::new())
}
@@ -1167,6 +1315,21 @@ impl StateHandler for CreateTopicWithAssignmentsRequest {
let new_revision = state.revision.wrapping_add(1);
state.revision = new_revision;
+ // Share one `Arc<TopicStats>` across both left-right buffers via the
+ // registry, parented to the stream's shared `Arc<StreamStats>`. The id
+ // the next insert will use is deterministic across buffers. Fetched
+ // under an immutable borrow that ends before the `&mut stream` below,
+ // so the registry access (a sibling field) does not alias.
+ let (topic_id, parent_stats) = {
+ let Some(stream) = state.items.get(stream_id) else {
+ return ApplyReply::err(CreateTopicResult::StreamNotFound);
+ };
+ (stream.topics.vacant_key(), stream.stats.clone())
+ };
+ let topic_stats = state
+ .stats_registry
+ .topic(stream_id, topic_id, parent_stats);
+
let Some(stream) = state.items.get_mut(stream_id) else {
return ApplyReply::err(CreateTopicResult::StreamNotFound);
};
@@ -1178,7 +1341,7 @@ impl StateHandler for CreateTopicWithAssignmentsRequest {
};
let topic = Topic {
- id: 0,
+ id: topic_id,
name: name_arc.clone(),
created_at: timestamp,
replication_factor,
@@ -1188,7 +1351,7 @@ impl StateHandler for CreateTopicWithAssignmentsRequest {
)
.unwrap_or_default(),
max_topic_size: MaxTopicSize::from(self.request.max_topic_size),
- stats: Arc::new(TopicStats::new(stream.stats.clone())),
+ stats: topic_stats,
partitions: Vec::new(),
round_robin_counter: Arc::new(AtomicUsize::new(0)),
consumer_groups: AHashMap::default(),
@@ -1196,10 +1359,9 @@ impl StateHandler for CreateTopicWithAssignmentsRequest {
next_consumer_group_id: 1,
};
- let topic_id = stream.topics.insert(topic);
- if let Some(topic) = stream.topics.get_mut(topic_id) {
- topic.id = topic_id;
-
+ let inserted = stream.topics.insert(topic);
+ debug_assert_eq!(inserted, topic_id, "vacant_key must match the insert slot");
+ if let Some(topic) = stream.topics.get_mut(inserted) {
for partition in &self.partitions {
let partition = Partition {
id: partition.partition_id as usize,
@@ -1339,6 +1501,8 @@ impl StateHandler for DeleteTopicRequest {
let name = topic.name.clone();
stream.topics.remove(topic_id);
stream.topic_index.remove(&name);
+ // Evict registry entry so a reused slab id starts with fresh stats.
+ state.stats_registry.remove_topic(stream_id, topic_id);
state.revision = state.revision.wrapping_add(1);
// The dropped topic may have held groups with pending revocations.
state.recompute_pending_revocations_count();
@@ -1476,11 +1640,15 @@ impl StateHandler for DeletePartitionsRequest {
let count_to_delete = self.partitions_count as usize;
let did_delete = count_to_delete > 0 && count_to_delete <= topic.partitions.len();
if did_delete {
- topic
- .partitions
- .truncate(topic.partitions.len() - count_to_delete);
+ let retained = topic.partitions.len() - count_to_delete;
+ topic.partitions.truncate(retained);
// Members assigned the removed partitions must give them up.
topic.rebalance_consumer_groups();
+ // Evict registry entries so re-created partition ids start with
+ // fresh stats.
+ state
+ .stats_registry
+ .remove_partitions_from(stream_id, topic_id, retained);
}
if did_delete {
state.revision = state.revision.wrapping_add(1);
@@ -1581,9 +1749,13 @@ impl Snapshotable for Streams {
) -> Result<Self, crate::stm::snapshot::SnapshotError> {
let mut index: AHashMap<Arc<str>, usize> = AHashMap::new();
let mut stream_entries: Vec<(usize, Stream)> = Vec::new();
+ // Register restored stats in the shared registry so both left-right
+ // buffers (and any post-restore op) reference one `Arc` per
+ // stream/topic (see `StatsRegistry`).
+ let stats_registry = Arc::new(StatsRegistry::default());
for (slab_key, stream_snap) in snapshot.items {
- let stream_stats = Arc::new(StreamStats::default());
+ let stream_stats = stats_registry.stream(slab_key);
stream_stats.store_from_snapshot(
stream_snap.stats.size_bytes,
stream_snap.stats.messages_count,
@@ -1594,7 +1766,8 @@ impl Snapshotable for Streams {
let mut topic_entries: Vec<(usize, Topic)> = Vec::new();
for (topic_slab_key, topic_snap) in stream_snap.topics {
- let topic_stats = Arc::new(TopicStats::new(stream_stats.clone()));
+ let topic_stats =
+ stats_registry.topic(slab_key, topic_slab_key, stream_stats.clone());
topic_stats.store_from_snapshot(
topic_snap.stats.size_bytes,
topic_snap.stats.messages_count,
@@ -1674,6 +1847,7 @@ impl Snapshotable for Streams {
// Recomputed from the restored groups just below.
pending_revocations_count: 0,
last_result: None,
+ stats_registry,
};
inner.recompute_pending_revocations_count();
Ok(inner.into())
diff --git a/core/partitions/src/iggy_index_reader.rs b/core/partitions/src/iggy_index_reader.rs
new file mode 100644
index 000000000..a105a6982
--- /dev/null
+++ b/core/partitions/src/iggy_index_reader.rs
@@ -0,0 +1,117 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::iggy_index::{IGGY_INDEX_SIZE, IggyIndex};
+use bytes::Buf;
+use compio::fs::{File, OpenOptions};
+use compio::io::AsyncReadAtExt;
+use iggy_common::IggyError;
+use tracing::trace;
+
+/// Reader for the sparse index file written by [`crate::IggyIndexWriter`].
+///
+/// The on-disk stride is [`IGGY_INDEX_SIZE`] (24 bytes: `offset` u64,
+/// `timestamp` u64, `position` u64, little-endian) — distinct from the legacy
+/// 16-byte dense per-message index that `server_common::IndexReader` parses.
+/// Recovery reaches for this reader so the reader matches the writer.
+#[derive(Debug)]
+pub struct IggyIndexReader {
+ file_path: String,
+ file: File,
+}
+
+impl IggyIndexReader {
+ /// Opens the sparse index file at `file_path` for reading.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if the file cannot be opened.
+ pub async fn new(file_path: &str) -> Result<Self, IggyError> {
+ let file = OpenOptions::new()
+ .read(true)
+ .open(file_path)
+ .await
+ .map_err(|_| IggyError::CannotReadFile)?;
+ Ok(Self {
+ file_path: file_path.to_owned(),
+ file,
+ })
+ }
+
+ /// Number of whole 24-byte entries in the file. A trailing partial entry
+ /// (torn write) is truncated by the integer division and ignored.
+ async fn entry_count(&self) -> Result<u64, IggyError> {
+ let size = self
+ .file
+ .metadata()
+ .await
+ .map_err(|_| IggyError::CannotReadFileMetadata)?
+ .len();
+ Ok(size / IGGY_INDEX_SIZE as u64)
+ }
+
+ async fn read_entry_at(&self, byte_position: u64) -> Result<IggyIndex, IggyError> {
+ // `with_capacity` (len 0): `read_exact_at` fills the spare capacity in
+ // place and advances the length, so a `vec![0u8; N]` (no spare) would
+ // read nothing.
+ let buffer = Vec::with_capacity(IGGY_INDEX_SIZE);
+ let (result, buffer): (std::io::Result<()>, Vec<u8>) =
+ self.file.read_exact_at(buffer, byte_position).await.into();
+ result.map_err(|_| IggyError::CannotReadFile)?;
+ let mut view = buffer.as_slice();
+ let offset = view.get_u64_le();
+ let timestamp = view.get_u64_le();
+ let position = view.get_u64_le();
+ trace!(
+ target: "iggy.partitions.storage",
+ file = self.file_path.as_str(),
+ offset,
+ timestamp,
+ position,
+ "read sparse index entry"
+ );
+ Ok(IggyIndex::new(offset, timestamp, position))
+ }
+
+ /// First index entry, or `None` when the file holds no whole entry.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if the file metadata or bytes cannot be read.
+ pub async fn load_first(&self) -> Result<Option<IggyIndex>, IggyError> {
+ if self.entry_count().await? == 0 {
+ return Ok(None);
+ }
+ Ok(Some(self.read_entry_at(0).await?))
+ }
+
+ /// Last index entry, or `None` when the file holds no whole entry.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if the file metadata or bytes cannot be read.
+ pub async fn load_last(&self) -> Result<Option<IggyIndex>, IggyError> {
+ let count = self.entry_count().await?;
+ if count == 0 {
+ return Ok(None);
+ }
+ Ok(Some(
+ self.read_entry_at((count - 1) * IGGY_INDEX_SIZE as u64)
+ .await?,
+ ))
+ }
+}
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index c396eb948..b3c5098af 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -33,10 +33,10 @@ use crate::{
use consensus::{
CommitLogEvent, Consensus, PartitionDiagEvent, Pipeline, PipelineEntry, PlaneKind, Project,
ReplicaLogContext, RequestLogEvent, Sequencer, SimEventKind, VsrConsensus, ack_preflight,
- ack_quorum_reached, build_reply_from_request, build_reply_message, drain_committable_prefix,
- emit_namespace_progress_event, emit_partition_diag, emit_sim_event,
- fence_old_prepare_by_commit, replicate_preflight, replicate_to_next_in_chain,
- send_prepare_ok as send_prepare_ok_common,
+ ack_quorum_reached, build_reply_from_request, build_reply_message,
+ build_result_rejection_reply, drain_committable_prefix, emit_namespace_progress_event,
+ emit_partition_diag, emit_sim_event, fence_old_prepare_by_commit, replicate_preflight,
+ replicate_to_next_in_chain, send_prepare_ok as send_prepare_ok_common,
};
use iggy_binary_protocol::requests::consumer_offsets::{
DeleteConsumerOffset2Request, DeleteConsumerOffsetRequest, StoreConsumerOffset2Request,
@@ -277,20 +277,20 @@ where
/// Stage a consumer offset delete for the replicated op. See
/// [`stage_consumer_offset_upsert`] for the ordering contract.
///
- /// # Errors
- ///
- /// Returns `ConsumerOffsetNotFound` if the consumer or group has no
- /// existing on-disk / in-memory offset to delete.
+ /// Deliberately infallible: this runs on the replicated-apply path (every
+ /// replica), where the offset may legitimately be absent (e.g. a backup
+ /// that never observed the primary-only `NoAck` store). The client-facing
+ /// "offset must exist" precondition is enforced once at primary admission
+ /// (`ensure_consumer_offset_exists` in `on_request`); re-checking here would
+ /// fail the replicated apply on such a replica and wedge the group.
pub(crate) fn stage_consumer_offset_delete(
&mut self,
op: u64,
kind: ConsumerKind,
consumer_id: u32,
- ) -> Result<(), IggyError> {
- self.ensure_consumer_offset_exists(kind, consumer_id)?;
+ ) {
let pending = PendingConsumerOffsetCommit::delete(kind, consumer_id);
self.pending_consumer_offset_commits.insert(op, pending);
- Ok(())
}
pub(crate) async fn apply_staged_consumer_offset_commit(
@@ -370,13 +370,25 @@ where
});
Ok(())
}
+ // Commit-time apply keeps its invariant check on the PRIMARY:
+ // admission verified the offset exists there, so a miss on the
+ // primary is real divergence (log corruption / out-of-order apply)
+ // and must surface rather than silently mask a split state. A
+ // FOLLOWER may legitimately miss the offset: `AckLevel::NoAck`
+ // (v2) stores apply on the primary only and are never replicated,
+ // so a later quorum delete finds nothing on the backups -- erroring
+ // there would fail the committed apply, panic the replica as
+ // divergent, and crash-loop on every journal replay. The
+ // prepare-time race is handled by not re-checking existence at
+ // staging (see `stage_consumer_offset_delete`).
PendingConsumerOffsetMutation::Delete if pending.kind == ConsumerKind::Consumer => {
let id = pending.consumer_id;
let guard = self.consumer_offsets.pin();
let key = usize::try_from(id).expect("u32 consumer id must fit usize");
- let _ = guard
- .remove(&key)
- .ok_or(IggyError::ConsumerOffsetNotFound(key))?;
+ let removed = guard.remove(&key).is_some();
+ if !removed && !self.consensus.is_follower() {
+ return Err(IggyError::ConsumerOffsetNotFound(key));
+ }
Ok(())
}
PendingConsumerOffsetMutation::Delete
@@ -387,9 +399,10 @@ where
let key = ConsumerGroupId(
usize::try_from(group_id).expect("u32 group id must fit usize"),
);
- let _ = guard
- .remove(&key)
- .ok_or(IggyError::ConsumerOffsetNotFound(key.0))?;
+ let removed = guard.remove(&key).is_some();
+ if !removed && !self.consensus.is_follower() {
+ return Err(IggyError::ConsumerOffsetNotFound(key.0));
+ }
Ok(())
}
_ => Ok(()),
@@ -496,7 +509,11 @@ where
return;
}
- let reply = build_reply_from_request(&self.consensus, &request_header, bytes::Bytes::new());
+ let reply = build_reply_from_request(
+ &self.consensus,
+ &request_header,
+ committed_reply_body(request_header.operation),
+ );
let reply_buffers = reply.into_generic().into_frozen();
if let Err(error) = self
.consensus
@@ -920,16 +937,24 @@ where
(segment_index, position)
}
MessageLookup::Timestamp { timestamp, .. } => {
- for (segment_index, _) in segments.iter().enumerate() {
- if let Some(index) = self
- .log
- .segment_indexes(segment_index)
- .and_then(|cache| cache.timestamp_lower_bound(*timestamp))
- {
- return (segment_index, index.position);
- }
- }
- (0, 0)
+ // Resolve the starting SEGMENT from segment metadata, not from
+ // the per-segment index caches: sealed segments drop their
+ // cache at rotation, and a cache miss must not read as "the
+ // timestamp is not in this segment" (skipping a sealed segment
+ // loses its messages). Timestamps are monotone across
+ // segments, so the first segment whose max timestamp reaches
+ // the query is the correct start; the walk filters precisely,
+ // so an early start is safe.
+ let segment_index = segments
+ .iter()
+ .position(|segment| segment.max_timestamp >= *timestamp)
+ .unwrap_or_else(|| segments.len().saturating_sub(1));
+ let position = self
+ .log
+ .segment_indexes(segment_index)
+ .and_then(|cache| cache.timestamp_lower_bound(*timestamp))
+ .map_or(0, |index| index.position);
+ (segment_index, position)
}
}
}
@@ -1027,9 +1052,62 @@ where
.with_operation(message.header().operation)
.with_error(error.to_string()),
);
+ // Reply with a terminal error instead of dropping the request:
+ // the SDK decodes any non-`TransientNotCommitted` result code as a
+ // final error, so the client gets `ConsumerOffsetNotFound` at once
+ // rather than replaying until its read-timeout (a client hang).
+ // The op is NOT admitted/replicated, so no replica ever applies a
+ // delete for an offset the primary knows is absent.
+ let reply = build_result_rejection_reply(
+ message.header(),
+ self.consensus().commit_max(),
+ error.as_code(),
+ );
+ let _ = self
+ .consensus()
+ .message_bus()
+ .send_to_client(client_id, reply.into_generic().into_frozen())
+ .await;
return;
}
+ // Reject an out-of-range consumer-offset store at admission with a
+ // terminal `InvalidOffset` (result-framed like the delete rejection),
+ // mirroring the legacy `validate_partition_offset`: an empty partition
+ // accepts no offset, and a stored offset may not run ahead of the
+ // committed offset. Done here so the doomed op is never replicated.
+ if matches!(
+ message.header().operation,
+ Operation::StoreConsumerOffset | Operation::StoreConsumerOffset2
+ ) && let Some((_, _, Some(requested_offset), _)) = consumer_offset
+ {
+ let current_offset = self.stats.current_offset();
+ let partition_empty =
+ self.stats.messages_count_inconsistent() == 0 && current_offset == 0;
+ if partition_empty || requested_offset > current_offset {
+ emit_partition_diag(
+ tracing::Level::WARN,
+ &PartitionDiagEvent::new(
+ ReplicaLogContext::from_consensus(consensus, PlaneKind::Partitions),
+ "rejecting store_consumer_offset for out-of-range offset",
+ )
+ .with_operation(message.header().operation)
+ .with_error(IggyError::InvalidOffset(requested_offset).to_string()),
+ );
+ let reply = build_result_rejection_reply(
+ message.header(),
+ self.consensus().commit_max(),
+ IggyError::InvalidOffset(requested_offset).as_code(),
+ );
+ let _ = self
+ .consensus()
+ .message_bus()
+ .send_to_client(client_id, reply.into_generic().into_frozen())
+ .await;
+ return;
+ }
+ }
+
assert!(!consensus.is_follower(), "on_request: primary only");
assert!(consensus.is_normal(), "on_request: status must be normal");
assert!(!consensus.is_syncing(), "on_request: must not be syncing");
@@ -1503,7 +1581,7 @@ where
);
}
Operation::DeleteConsumerOffset | Operation::DeleteConsumerOffset2 => {
- self.stage_consumer_offset_delete(header.op, kind, consumer_id)?;
+ self.stage_consumer_offset_delete(header.op, kind, consumer_id);
}
_ => unreachable!(),
}
@@ -1701,11 +1779,10 @@ where
self.rotate_segment(config).await?;
}
- self.stats
- .increment_size_bytes(committed_info.size.as_bytes_u64());
- self.stats
- .increment_messages_count(u64::from(committed_info.messages_count));
-
+ // Aggregate stats (`messages_count`/`size_bytes`) advance at commit in
+ // `commit_partition_entry`, not here: this persist path is threshold-
+ // gated, so counting here would leave the stats lagging the visible
+ // offset until a flush and would double-count once it fires.
let durable_offset = committed_info.current_offset;
self.offset.store(durable_offset, Ordering::Release);
self.stats.set_current_offset(durable_offset);
@@ -1812,7 +1889,10 @@ where
// No reply cache: at-least-once means retries re-commit at new
// offsets. Only primary delivers replies; backups just advance
// commit. Session lifecycle is metadata-only.
- let reply = build_reply_message(&prepare_header, &bytes::Bytes::new());
+ let reply = build_reply_message(
+ &prepare_header,
+ &committed_reply_body(prepare_header.operation),
+ );
// TODO: no production caller yet. Partition has no in-process
// subscriber (only metadata uses pipeline_message_with_subscriber);
@@ -1863,7 +1943,7 @@ where
async fn resolve_committed_visible_offsets(
&self,
drained: &[PipelineEntry],
- ) -> HashMap<u64, u64> {
+ ) -> HashMap<u64, CommittedBatchStats> {
let mut committed_visible_offsets = HashMap::new();
for entry in drained {
@@ -1871,9 +1951,9 @@ where
continue;
}
- match self.committed_end_offset_for_prepare(&entry.header).await {
- Ok(Some(end_offset)) => {
- committed_visible_offsets.insert(entry.header.op, end_offset);
+ match self.committed_batch_stats_for_prepare(&entry.header).await {
+ Ok(Some(batch_stats)) => {
+ committed_visible_offsets.insert(entry.header.op, batch_stats);
}
Ok(None) => {}
Err(error) => {
@@ -1898,7 +1978,7 @@ where
&mut self,
prepare_header: PrepareHeader,
messages_committed: &mut bool,
- committed_visible_offsets: &HashMap<u64, u64>,
+ committed_visible_offsets: &HashMap<u64, CommittedBatchStats>,
failed_commit: &mut bool,
config: &PartitionsConfig,
) -> bool {
@@ -1922,9 +2002,16 @@ where
*messages_committed = true;
}
- if let Some(visible_offset) = committed_visible_offsets.get(&prepare_header.op) {
- self.offset.store(*visible_offset, Ordering::Release);
- self.stats.set_current_offset(*visible_offset);
+ if let Some(batch_stats) = committed_visible_offsets.get(&prepare_header.op) {
+ self.offset.store(batch_stats.end_offset, Ordering::Release);
+ self.stats.set_current_offset(batch_stats.end_offset);
+ // Advance the aggregate stats with the visible offset. Disk
+ // persistence is threshold-gated in `commit_messages`, which
+ // must not also touch these counters or committed messages
+ // would be double-counted once they flush.
+ self.stats
+ .increment_messages_count(u64::from(batch_stats.message_count));
+ self.stats.increment_size_bytes(batch_stats.size_bytes);
}
!*failed_commit
}
@@ -1950,10 +2037,10 @@ where
}
}
- async fn committed_end_offset_for_prepare(
+ async fn committed_batch_stats_for_prepare(
&self,
prepare_header: &PrepareHeader,
- ) -> Result<Option<u64>, IggyError> {
+ ) -> Result<Option<CommittedBatchStats>, IggyError> {
let Some(entry) = self.log.journal().inner.entry(prepare_header).await else {
return Err(IggyError::InvalidCommand);
};
@@ -1964,9 +2051,11 @@ where
return Ok(None);
}
- Ok(Some(
- batch.header.base_offset + u64::from(message_count) - 1,
- ))
+ Ok(Some(CommittedBatchStats {
+ end_offset: batch.header.base_offset + u64::from(message_count) - 1,
+ message_count,
+ size_bytes: batch.header.total_size() as u64,
+ }))
}
fn parse_consumer_offset_request(
@@ -2253,6 +2342,12 @@ where
let _ = old_storage.shutdown();
self.log.messages_writers_mut()[old_segment_index] = None;
self.log.index_writers_mut()[old_segment_index] = None;
+ // Drop the sealed segment's in-memory index cache: only the ACTIVE
+ // segment's cache is ever read (the `commit_messages` flush staging),
+ // so a sealed cache is dead weight -- and `ensure_indexes` preallocates
+ // a 16 MiB-capacity `Vec` per segment, which under small-segment
+ // workloads retains hundreds of MB across thousands of sealed segments.
+ self.log.indexes_mut()[old_segment_index] = None;
self.log
.add_persisted_segment(segment, storage, Some(messages_writer), Some(index_writer));
@@ -2632,6 +2727,29 @@ fn peek_operation(entry: &Frozen<4096>) -> Operation {
.operation
}
+/// Success reply body for a committed partition op. Result-framed ops
+/// (`Operation::is_result_framed`; on this plane the consumer-offset ops,
+/// whose rejections ship typed errors) must carry an explicit empty result
+/// section (`[count = 0]`) so the SDK's framed decode does not misread the
+/// payload; every other partition op replies with an empty body.
+const fn committed_reply_body(operation: Operation) -> bytes::Bytes {
+ if operation.is_result_framed() {
+ bytes::Bytes::from_static(&[0, 0, 0, 0])
+ } else {
+ bytes::Bytes::new()
+ }
+}
+
+/// Committed-batch accounting surfaced at commit time so the aggregate stats
+/// (`messages_count`, `size_bytes`) advance with the visible offset rather than
+/// waiting on the threshold-gated disk persist.
+#[derive(Clone, Copy)]
+struct CommittedBatchStats {
+ end_offset: u64,
+ message_count: u32,
+ size_bytes: u64,
+}
+
/// Fold one `SendMessages` batch's accounting into a running `JournalInfo`,
/// matching the field updates `append_messages` applies per append.
/// `current_offset` is the batch's last message offset; the batch carries a
diff --git a/core/partitions/src/lib.rs b/core/partitions/src/lib.rs
index 3942286e7..12ba1cde1 100644
--- a/core/partitions/src/lib.rs
+++ b/core/partitions/src/lib.rs
@@ -18,6 +18,7 @@
#![allow(clippy::future_not_send)]
mod iggy_index;
+mod iggy_index_reader;
mod iggy_index_writer;
mod iggy_partition;
mod iggy_partitions;
@@ -31,6 +32,8 @@ mod types;
use iggy_binary_protocol::PrepareHeader;
use iggy_common::IggyError;
+pub use iggy_index::IggyIndex;
+pub use iggy_index_reader::IggyIndexReader;
pub use iggy_index_writer::IggyIndexWriter;
pub use iggy_partition::IggyPartition;
pub use iggy_partitions::IggyPartitions;
diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs
index a1e1360fc..8b41ac790 100644
--- a/core/sdk/src/quic/quic_client.rs
+++ b/core/sdk/src/quic/quic_client.rs
@@ -54,6 +54,23 @@ const REQUEST_INITIAL_BYTES_LENGTH: usize = 4;
const RESPONSE_INITIAL_BYTES_LENGTH: usize = 8;
const NAME: &str = "Iggy";
+/// Bound on how long a single QUIC request waits for its response, mirroring the
+/// TCP client's `RESPONSE_READ_TIMEOUT`. A replicated request that the server
+/// cannot commit transiently is answered with silence - the server expects the
+/// SDK read-timeout to drive a replay. `RecvStream::read_to_end` on an
+/// unanswered bidi stream otherwise blocks until the QUIC idle timeout (which
+/// can be minutes), parking that request and, because the connection is held
+/// under lock per send, every later request on the client.
+const RESPONSE_READ_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// Backoff before replaying a request the server answered with an explicit
+/// `TransientNotCommitted` frame (not-caught-up / in-flight / pipeline-full /
+/// view-change cancel). Unlike a silent timeout, this reply arrives promptly, so
+/// a short pause keeps the replay from spinning while the primary catches up.
+/// Bounded overall by `RESPONSE_READ_TIMEOUT`.
+#[cfg(feature = "vsr")]
+const NOT_READY_RETRY_INTERVAL: Duration = Duration::from_millis(50);
+
/// QUIC client for interacting with the Iggy API.
#[derive(Debug)]
pub struct QuicClient {
@@ -144,7 +161,13 @@ impl BinaryTransport for QuicClient {
}
#[cfg(feature = "vsr")]
- if matches!(self.config.auto_login, AutoLogin::Disabled) {
+ if matches!(self.config.auto_login, AutoLogin::Disabled) && !is_login_register_code(code) {
+ // Without auto-login a reconnect cannot re-establish the session, so
+ // non-login requests are not recovered here - their transient replay
+ // happens on the live connection inside `send_raw`. Login/register
+ // is the exception: the server stays deliberately silent on a
+ // transient register failure and relies on the client replaying via
+ // a reconnect with a fresh session.
return Err(error);
}
@@ -154,6 +177,13 @@ impl BinaryTransport for QuicClient {
#[cfg(feature = "vsr")]
if skip_auto_login {
*self.skip_auto_login_once.lock().await = true;
+ // The replayed login/register must mint a fresh Register: the failed
+ // attempt already consumed the one-shot register request id and may
+ // have half-bound the session.
+ *self
+ .consensus_session
+ .lock()
+ .expect("consensus session mutex poisoned") = ConsensusSession::new();
}
let server_address = self.current_server_address.lock().await.to_string();
info!(
@@ -302,10 +332,14 @@ impl QuicClient {
async fn handle_response(
recv: &mut RecvStream,
response_buffer_size: usize,
+ read_timeout: Duration,
) -> Result<Bytes, IggyError> {
- let buffer = recv
- .read_to_end(response_buffer_size)
+ let buffer = tokio::time::timeout(read_timeout, recv.read_to_end(response_buffer_size))
.await
+ .map_err(|_| {
+ error!("Timed out after {read_timeout:?} waiting for QUIC response");
+ IggyError::Disconnected
+ })?
.map_err(|error| {
error!("Failed to read response data: {error}");
IggyError::QuicError
@@ -660,73 +694,127 @@ impl QuicClient {
// SAFETY: we run code holding the `connection` lock in a task so we can't be cancelled while holding the lock.
tokio::spawn(async move {
let connection = connection.lock().await;
- if let Some(connection) = connection.as_ref() {
- #[cfg(feature = "vsr")]
+ let Some(connection) = connection.as_ref() else {
+ error!("Cannot send data. Client is not connected.");
+ return Err(IggyError::NotConnected);
+ };
+
+ #[cfg(feature = "vsr")]
+ {
let (request_header, request_size) = {
let mut consensus_session = consensus_session
.lock()
.expect("consensus session mutex poisoned");
crate::vsr::encode_request_header(&mut consensus_session, code, &payload)?
};
- #[cfg(not(feature = "vsr"))]
- let payload_length = payload.len() + REQUEST_INITIAL_BYTES_LENGTH;
- let (mut send, mut recv) = connection.open_bi().await.map_err(|error| {
- error!("Failed to open a bidirectional stream: {error}");
- IggyError::QuicError
- })?;
- trace!("Sending a QUIC request with code: {code}");
- #[cfg(feature = "vsr")]
- trace!(
- "Sending a QUIC VSR request of size {} with code: {code}",
- request_size
- );
- #[cfg(feature = "vsr")]
- send.write_all(bytemuck::bytes_of(&request_header))
- .await
- .map_err(|error| {
+ trace!("Sending a QUIC VSR request of size {request_size} with code: {code}");
+ // Same-connection transient resend, gated on the EXPLICIT
+ // `TransientNotCommitted` frame only. The server answers every
+ // transient submit with that frame (it is pre-commit by
+ // construction, so replaying the SAME request header on a fresh
+ // bidi cannot double-commit), and it no longer abandons a bidi
+ // whose op is still committing. Silence therefore is NOT a
+ // retry signal: partition ops share one request id and have no
+ // reply cache, so resending a silently-unanswered request whose
+ // first attempt was buffered and later commits would commit it
+ // twice (duplicate `SendMessages`, or a succeeded delete coming
+ // back as terminal `ConsumerOffsetNotFound`). A silent deadline
+ // expiry surfaces `Disconnected` and takes the
+ // reconnect path in `send_raw_with_response`, same as TCP.
+ let header_bytes = bytemuck::bytes_of(&request_header);
+ let deadline = tokio::time::Instant::now() + RESPONSE_READ_TIMEOUT;
+ loop {
+ let (mut send, mut recv) = connection.open_bi().await.map_err(|error| {
+ error!("Failed to open a bidirectional stream: {error}");
+ IggyError::QuicError
+ })?;
+ send.write_all(header_bytes).await.map_err(|error| {
error!("Failed to write VSR request header: {error}");
IggyError::QuicError
})?;
- #[cfg(feature = "vsr")]
- if !payload.is_empty() {
- send.write_all(&payload).await.map_err(|error| {
- error!("Failed to write VSR request payload: {error}");
+ if !payload.is_empty() {
+ send.write_all(&payload).await.map_err(|error| {
+ error!("Failed to write VSR request payload: {error}");
+ IggyError::QuicError
+ })?;
+ }
+ send.finish().map_err(|error| {
+ error!("Failed to finish VSR request stream: {error}");
IggyError::QuicError
})?;
+ let remaining =
+ deadline.saturating_duration_since(tokio::time::Instant::now());
+ if remaining.is_zero() {
+ return Err(IggyError::Disconnected);
+ }
+ match QuicClient::handle_response(
+ &mut recv,
+ response_buffer_size as usize,
+ remaining,
+ )
+ .await
+ {
+ Ok(reply) => return Ok(reply),
+ // `TransientNotCommitted` = the server replied with an
+ // explicit retry frame because it could not commit yet
+ // (not-caught-up / in-flight / pipeline-full /
+ // view-change cancel). Nothing committed, so replaying
+ // the same request id on a fresh bidi is safe; the
+ // session stays intact so no reconnect/relogin is
+ // needed (login replays idempotently too). Anything
+ // else - including a silent read timeout - is
+ // terminal here and handled by the caller.
+ Err(IggyError::TransientNotCommitted)
+ if tokio::time::Instant::now() < deadline =>
+ {
+ // The explicit frame returns promptly (no read
+ // timeout elapsed), so pace the replay.
+ let remaining =
+ deadline.saturating_duration_since(tokio::time::Instant::now());
+ tokio::time::sleep(NOT_READY_RETRY_INTERVAL.min(remaining)).await;
+ warn!(
+ "QUIC request code {code} not committed (transient); resending on a new stream"
+ );
+ }
+ Err(error) => return Err(error),
+ }
}
- #[cfg(feature = "vsr")]
- send.finish().map_err(|error| {
- error!("Failed to finish VSR request stream: {error}");
+ }
+
+ #[cfg(not(feature = "vsr"))]
+ {
+ let payload_length = payload.len() + REQUEST_INITIAL_BYTES_LENGTH;
+ let (mut send, mut recv) = connection.open_bi().await.map_err(|error| {
+ error!("Failed to open a bidirectional stream: {error}");
IggyError::QuicError
})?;
- #[cfg(not(feature = "vsr"))]
+ trace!("Sending a QUIC request with code: {code}");
send.write_all(&(payload_length as u32).to_le_bytes())
.await
.map_err(|error| {
error!("Failed to write payload length: {error}");
IggyError::QuicError
})?;
- #[cfg(not(feature = "vsr"))]
send.write_all(&code.to_le_bytes()).await.map_err(|error| {
error!("Failed to write payload code: {error}");
IggyError::QuicError
})?;
- #[cfg(not(feature = "vsr"))]
send.write_all(&payload).await.map_err(|error| {
error!("Failed to write payload: {error}");
IggyError::QuicError
})?;
- #[cfg(not(feature = "vsr"))]
send.finish().map_err(|error| {
error!("Failed to finish sending data: {error}");
IggyError::QuicError
})?;
trace!("Sent a QUIC request with code: {code}, waiting for a response...");
- return QuicClient::handle_response(&mut recv, response_buffer_size as usize).await;
+ QuicClient::handle_response(
+ &mut recv,
+ response_buffer_size as usize,
+ RESPONSE_READ_TIMEOUT,
+ )
+ .await
}
-
- error!("Cannot send data. Client is not connected.");
- Err(IggyError::NotConnected)
})
.await
.map_err(|e| {
diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs
index 127a2f58f..d69ffd2ae 100644
--- a/core/sdk/src/tcp/tcp_client.rs
+++ b/core/sdk/src/tcp/tcp_client.rs
@@ -64,6 +64,14 @@ const NAME: &str = "Iggy";
#[cfg(feature = "vsr")]
const RESPONSE_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
+/// Backoff before replaying a request the server answered with an explicit
+/// `TransientNotCommitted` frame (not-caught-up / in-flight / pipeline-full /
+/// view-change cancel). The reply arrives promptly, so a short pause keeps the
+/// replay from spinning while the primary catches up. Bounded by
+/// `RESPONSE_READ_TIMEOUT`.
+#[cfg(feature = "vsr")]
+const NOT_READY_RETRY_INTERVAL: std::time::Duration = std::time::Duration::from_millis(50);
+
/// TCP client for interacting with the Iggy API.
/// It requires a valid server address.
#[derive(Debug)]
@@ -700,107 +708,122 @@ impl TcpClient {
let result = tokio::spawn(async move {
let mut stream = stream.lock().await;
if let Some(stream) = stream.as_mut() {
- #[cfg(feature = "vsr")]
- let (request_header, request_size) = {
- let mut consensus_session = consensus_session
- .lock()
- .expect("consensus session mutex poisoned");
- crate::vsr::encode_request_header(&mut consensus_session, code, &payload)?
- };
- #[cfg(not(feature = "vsr"))]
- let payload_length = payload.len() + REQUEST_INITIAL_BYTES_LENGTH;
- #[cfg(feature = "vsr")]
- trace!(
- "Sending a TCP VSR request of size {} with code: {code}",
- request_size
- );
- #[cfg(not(feature = "vsr"))]
- trace!("Sending a TCP request of size {payload_length} with code: {code}");
- #[cfg(feature = "vsr")]
- stream.write(bytemuck::bytes_of(&request_header)).await?;
- #[cfg(feature = "vsr")]
- if !payload.is_empty() {
- stream.write(&payload).await?;
- }
- #[cfg(not(feature = "vsr"))]
- stream.write(&(payload_length as u32).to_le_bytes()).await?;
- #[cfg(not(feature = "vsr"))]
- stream.write(&code.to_le_bytes()).await?;
- #[cfg(not(feature = "vsr"))]
- stream.write(&payload).await?;
- stream.flush().await?;
- trace!("Sent a TCP request with code: {code}, waiting for a response...");
#[cfg(feature = "vsr")]
{
- let mut response_header = [0u8; iggy_binary_protocol::HEADER_SIZE];
- // `stream.read` delegates to `read_exact`; on success it
- // always returns the requested length, so no short-read
- // guard is needed here.
- //
- // Deadline guards against server-side reply loss (e.g. a
- // stalled replication quorum that never commits the op):
- // the connection is lockstep, so an unanswered read would
- // 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.
- //
- // 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}",
- );
- return Err(IggyError::Disconnected);
+ // Encode the request header ONCE: `next_request_id` advances
+ // here, so a transient replay must reuse the same id for the
+ // server's dedup. The connection is lockstep (one request in
+ // flight per client), so a complete reply leaves the stream at
+ // a clean frame boundary -- a `TransientNotCommitted` answer
+ // (the server could not commit yet: not-caught-up / in-flight
+ // / pipeline-full / view-change cancel) lets us resend the
+ // SAME request on the SAME connection with no reconnect and
+ // the session intact. Bounded overall by RESPONSE_READ_TIMEOUT.
+ let (request_header, request_size) = {
+ let mut consensus_session = consensus_session
+ .lock()
+ .expect("consensus session mutex poisoned");
+ crate::vsr::encode_request_header(&mut consensus_session, code, &payload)?
};
- header_read.map_err(|error| {
- error!(
- "Failed to read VSR response header for TCP request with code: {code}: {error}",
- code = code,
- error = error
- );
- IggyError::Disconnected
- })?;
-
- let response_size = crate::vsr::response_size(&response_header)?;
-
- 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_at(
- response_deadline,
- stream.read_buf(&mut body, body_size),
+ trace!("Sending a TCP VSR request of size {request_size} with code: {code}");
+ let header_bytes = bytemuck::bytes_of(&request_header);
+ // One deadline bounds the whole request including transient
+ // replays, so a slow primary cannot stretch a single call
+ // beyond RESPONSE_READ_TIMEOUT.
+ let retry_deadline = tokio::time::Instant::now() + RESPONSE_READ_TIMEOUT;
+ loop {
+ stream.write(header_bytes).await?;
+ if !payload.is_empty() {
+ stream.write(&payload).await?;
+ }
+ stream.flush().await?;
+ trace!("Sent a TCP request with code: {code}, waiting for a response...");
+
+ let mut response_header = [0u8; iggy_binary_protocol::HEADER_SIZE];
+ // `stream.read` delegates to `read_exact`; on success it
+ // always returns the requested length, so no short-read
+ // guard is needed here.
+ //
+ // Deadline guards against server-side reply loss (e.g. a
+ // stalled replication quorum that never commits the op):
+ // the connection is lockstep, so an unanswered read would
+ // 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.
+ //
+ // 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.
+ let header_read = tokio::time::timeout_at(
+ retry_deadline,
+ stream.read(&mut response_header),
)
.await;
- let Ok(body_read) = body_read else {
+ let Ok(header_read) = header_read else {
error!(
- "Timed out after {RESPONSE_READ_TIMEOUT:?} waiting for VSR response body for TCP request with code: {code}",
+ "Timed out after {RESPONSE_READ_TIMEOUT:?} waiting for VSR response header for TCP request with code: {code}",
);
return Err(IggyError::Disconnected);
};
- body_read.map_err(|error| {
+ header_read.map_err(|error| {
error!(
- "Failed to read VSR response body for TCP request with code: {code}: {error}",
- code = code,
- error = error
+ "Failed to read VSR response header for TCP request with code: {code}: {error}",
);
IggyError::Disconnected
})?;
- body.freeze()
- } else {
- Bytes::new()
- };
- return crate::vsr::decode_response_split(&response_header, body);
+ let response_size = crate::vsr::response_size(&response_header)?;
+ 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_at(
+ retry_deadline,
+ stream.read_buf(&mut body, body_size),
+ )
+ .await;
+ let Ok(body_read) = body_read else {
+ error!(
+ "Timed out after {RESPONSE_READ_TIMEOUT:?} waiting for VSR response body for TCP request with code: {code}",
+ );
+ return Err(IggyError::Disconnected);
+ };
+ body_read.map_err(|error| {
+ error!(
+ "Failed to read VSR response body for TCP request with code: {code}: {error}",
+ );
+ IggyError::Disconnected
+ })?;
+ body.freeze()
+ } else {
+ Bytes::new()
+ };
+
+ match crate::vsr::decode_response_split(&response_header, body) {
+ // Server could not commit yet but answered with a
+ // complete frame; the lockstep stream is in sync, so
+ // replay the same request id on this connection after
+ // a short pause (no reconnect, session intact).
+ Err(IggyError::TransientNotCommitted)
+ if tokio::time::Instant::now() < retry_deadline =>
+ {
+ let remaining = retry_deadline
+ .saturating_duration_since(tokio::time::Instant::now());
+ tokio::time::sleep(NOT_READY_RETRY_INTERVAL.min(remaining)).await;
+ }
+ other => return other,
+ }
+ }
}
#[cfg(not(feature = "vsr"))]
{
+ let payload_length = payload.len() + REQUEST_INITIAL_BYTES_LENGTH;
+ trace!("Sending a TCP request of size {payload_length} with code: {code}");
+ stream.write(&(payload_length as u32).to_le_bytes()).await?;
+ stream.write(&code.to_le_bytes()).await?;
+ stream.write(&payload).await?;
+ stream.flush().await?;
+ trace!("Sent a TCP request with code: {code}, waiting for a response...");
let mut response_buffer = [0u8; RESPONSE_INITIAL_BYTES_LENGTH];
let read_bytes = stream.read(&mut response_buffer).await.map_err(|error| {
error!(
diff --git a/core/sdk/src/vsr.rs b/core/sdk/src/vsr.rs
index 8f7f45624..54fdb273c 100644
--- a/core/sdk/src/vsr.rs
+++ b/core/sdk/src/vsr.rs
@@ -243,7 +243,17 @@ fn read_operation(header_bytes: &[u8; HEADER_SIZE]) -> Result<Operation, IggyErr
/// section is corruption, never a silent success, so it maps to `InvalidCommand`
/// rather than risk a rejection decoding as `Ok`.
fn split_metadata_result(operation: Operation, body: Bytes) -> Result<Bytes, IggyError> {
- if !operation.is_metadata() {
+ // Register (login/register) replies are result-framed too, so a transient
+ // login decodes to `TransientNotCommitted` and the SDK replays it. The one
+ // exception is a terminal failure, which ships an empty body (no result
+ // section) and is passed through to fail the typed `LoginRegisterResponse`
+ // decode. `Operation::is_result_framed` is the shared source of truth with
+ // the server-side encode sites; the Register empty-body-is-terminal nuance
+ // is the one SDK-side addition. Other reads, data-plane ops, and Logout
+ // carry no result section and pass through untouched.
+ let result_framed =
+ operation.is_result_framed() || (operation == Operation::Register && !body.is_empty());
+ if !result_framed {
return Ok(body);
}
match result_code(&body) {
@@ -681,4 +691,74 @@ mod tests {
let out = split_metadata_result(Operation::NonReplicated, body.clone()).unwrap();
assert_eq!(out, body);
}
+
+ /// `[count = 1][index = 0][result = code]` -- the single-entry rejection
+ /// shape (`ApplyReply::write_reply_body` / `build_transient_reply`).
+ fn rejection_body(code: u32) -> Bytes {
+ let mut body = Vec::with_capacity(12);
+ body.extend_from_slice(&1u32.to_le_bytes());
+ body.extend_from_slice(&0u32.to_le_bytes());
+ body.extend_from_slice(&code.to_le_bytes());
+ Bytes::from(body)
+ }
+
+ /// `[count = 0]` followed by the payload -- the result-framed success shape.
+ fn success_body(payload: &[u8]) -> Bytes {
+ let mut body = Vec::with_capacity(4 + payload.len());
+ body.extend_from_slice(&0u32.to_le_bytes());
+ body.extend_from_slice(payload);
+ Bytes::from(body)
+ }
+
+ #[test]
+ fn delete_consumer_offset_rejection_decodes_to_terminal_error() {
+ let code = IggyError::ConsumerOffsetNotFound(0).as_code();
+ let result = split_metadata_result(Operation::DeleteConsumerOffset, rejection_body(code));
+ assert_eq!(
+ result.unwrap_err().as_code(),
+ code,
+ "delete rejection must surface as the typed error, not decode as Ok"
+ );
+ }
+
+ #[test]
+ fn store_consumer_offset_rejection_decodes_to_terminal_error() {
+ let code = IggyError::InvalidOffset(42).as_code();
+ let result = split_metadata_result(Operation::StoreConsumerOffset2, rejection_body(code));
+ assert_eq!(result.unwrap_err().as_code(), code);
+ }
+
+ #[test]
+ fn consumer_offset_success_strips_the_empty_result_section() {
+ let out = split_metadata_result(Operation::StoreConsumerOffset, success_body(b"")).unwrap();
+ assert!(out.is_empty());
+ let out =
+ split_metadata_result(Operation::DeleteConsumerOffset2, success_body(b"")).unwrap();
+ assert!(out.is_empty());
+ }
+
+ #[test]
+ fn metadata_transient_code_decodes_to_transient_not_committed() {
+ let code = IggyError::TransientNotCommitted.as_code();
+ let result = split_metadata_result(Operation::CreateStream, rejection_body(code));
+ assert!(matches!(
+ result.unwrap_err(),
+ IggyError::TransientNotCommitted
+ ));
+ }
+
+ #[test]
+ fn metadata_success_returns_payload_after_result_section() {
+ let out = split_metadata_result(Operation::CreateStream, success_body(b"payload")).unwrap();
+ assert_eq!(out.as_ref(), b"payload");
+ }
+
+ #[test]
+ fn send_messages_body_is_never_interpreted_as_a_result_section() {
+ // A data-plane reply whose first bytes happen to look like a rejection
+ // must pass through untouched: `SendMessages` is not result-framed.
+ let body = rejection_body(IggyError::InvalidOffset(1).as_code());
+ let out = split_metadata_result(Operation::SendMessages, body.clone()).unwrap();
+ assert_eq!(out, body);
+ }
}
diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs
index bbeee44e9..ab3786e11 100644
--- a/core/sdk/src/websocket/websocket_client.rs
+++ b/core/sdk/src/websocket/websocket_client.rs
@@ -66,6 +66,14 @@ const NAME: &str = "WebSocket";
#[cfg(feature = "vsr")]
const RESPONSE_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
+/// Backoff before replaying a request the server answered with an explicit
+/// `TransientNotCommitted` frame (not-caught-up / in-flight / pipeline-full /
+/// view-change cancel). The reply arrives promptly, so a short pause keeps the
+/// replay from spinning while the primary catches up. Bounded by
+/// `RESPONSE_READ_TIMEOUT`.
+#[cfg(feature = "vsr")]
+const NOT_READY_RETRY_INTERVAL: std::time::Duration = std::time::Duration::from_millis(50);
+
#[derive(Debug)]
pub struct WebSocketClient {
stream: Arc<Mutex<Option<WebSocketStreamKind>>>,
@@ -671,96 +679,111 @@ impl WebSocketClient {
}
let mut stream_guard = self.stream.lock().await;
- let stream = stream_guard.as_mut().ok_or_else(|| {
+ if stream_guard.is_none() {
trace!("Cannot send data. Client is not connected.");
- IggyError::NotConnected
- })?;
-
- #[cfg(feature = "vsr")]
- let request = {
- let mut consensus_session = self
- .consensus_session
- .lock()
- .expect("consensus session mutex poisoned");
- crate::vsr::encode_contiguous_request(&mut consensus_session, code, &payload)?
- };
- #[cfg(not(feature = "vsr"))]
- let payload_length = payload.len() + REQUEST_INITIAL_BYTES_LENGTH;
- #[cfg(not(feature = "vsr"))]
- let mut request = BytesMut::with_capacity(4 + REQUEST_INITIAL_BYTES_LENGTH + payload.len());
- #[cfg(not(feature = "vsr"))]
- request.put_u32_le(payload_length as u32);
- #[cfg(not(feature = "vsr"))]
- request.put_u32_le(code);
- #[cfg(not(feature = "vsr"))]
- request.put_slice(&payload);
-
- trace!(
- "Sending {NAME} message with code: {}, payload size: {} bytes",
- code,
- payload.len()
- );
- #[cfg(feature = "vsr")]
- trace!(
- "Sending {NAME} VSR request of size {} with code: {code}",
- request.len()
- );
-
- stream.write(&request).await?;
- stream.flush().await?;
+ return Err(IggyError::NotConnected);
+ }
#[cfg(feature = "vsr")]
{
- // Mirror the TCP path: header onto stack, body into its own
- // buffer, `decode_response_split` slices without concatenation.
- // 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];
- 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);
+ // Encode the request ONCE: `next_request_id` advances here, so a
+ // transient replay must reuse the same id for the server's dedup.
+ // The connection is lockstep (one request in flight per client), so a
+ // complete reply leaves the stream at a clean frame boundary -- a
+ // `TransientNotCommitted` answer (the server could not commit yet)
+ // lets us resend the SAME request on the SAME connection with no
+ // reconnect and the session intact. Bounded by RESPONSE_READ_TIMEOUT.
+ let request = {
+ let mut consensus_session = self
+ .consensus_session
+ .lock()
+ .expect("consensus session mutex poisoned");
+ crate::vsr::encode_contiguous_request(&mut consensus_session, code, &payload)?
};
- header_read?;
-
- let response_size = crate::vsr::response_size(&response_header)?;
- let body_size = response_size - iggy_binary_protocol::HEADER_SIZE;
- let body = if body_size > 0 {
- // `WebSocketStreamKind::read` reads into a slice without a
- // zero-fill prerequisite; we still allocate `body_size` but
- // skip the header concatenation.
- let mut body = vec![0u8; body_size];
- let body_read =
- tokio::time::timeout_at(response_deadline, stream.read(&mut body)).await;
- let Ok(body_read) = body_read else {
+ trace!(
+ "Sending {NAME} VSR request of size {} with code: {code}",
+ request.len()
+ );
+ // One deadline bounds the whole request including transient replays.
+ let retry_deadline = tokio::time::Instant::now() + RESPONSE_READ_TIMEOUT;
+ loop {
+ let stream = stream_guard.as_mut().ok_or(IggyError::NotConnected)?;
+ stream.write(&request).await?;
+ stream.flush().await?;
+
+ // 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 mut response_header = [0u8; iggy_binary_protocol::HEADER_SIZE];
+ let header_read =
+ tokio::time::timeout_at(retry_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 body for request with code: {code}"
+ "Timed out after {RESPONSE_READ_TIMEOUT:?} waiting for {NAME} VSR response header for request with code: {code}"
);
*stream_guard = None;
return Err(IggyError::Disconnected);
};
- body_read?;
- Bytes::from(body)
- } else {
- Bytes::new()
- };
+ header_read?;
+
+ let response_size = crate::vsr::response_size(&response_header)?;
+ let body_size = response_size - iggy_binary_protocol::HEADER_SIZE;
+ let body = if body_size > 0 {
+ let mut body = vec![0u8; body_size];
+ let body_read =
+ tokio::time::timeout_at(retry_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()
+ };
- crate::vsr::decode_response_split(&response_header, body)
+ match crate::vsr::decode_response_split(&response_header, body) {
+ // Server could not commit yet but answered with a complete
+ // frame; the lockstep stream is in sync, so replay the same
+ // request id on this connection after a short pause.
+ Err(IggyError::TransientNotCommitted)
+ if tokio::time::Instant::now() < retry_deadline =>
+ {
+ let remaining =
+ retry_deadline.saturating_duration_since(tokio::time::Instant::now());
+ tokio::time::sleep(NOT_READY_RETRY_INTERVAL.min(remaining)).await;
+ }
+ other => return other,
+ }
+ }
}
#[cfg(not(feature = "vsr"))]
{
+ let stream = stream_guard.as_mut().ok_or_else(|| {
+ trace!("Cannot send data. Client is not connected.");
+ IggyError::NotConnected
+ })?;
+ let payload_length = payload.len() + REQUEST_INITIAL_BYTES_LENGTH;
+ let mut request =
+ BytesMut::with_capacity(4 + REQUEST_INITIAL_BYTES_LENGTH + payload.len());
+ request.put_u32_le(payload_length as u32);
+ request.put_u32_le(code);
+ request.put_slice(&payload);
+ trace!(
+ "Sending {NAME} message with code: {}, payload size: {} bytes",
+ code,
+ payload.len()
+ );
+ stream.write(&request).await?;
+ stream.flush().await?;
+
let mut response_initial_buffer = vec![0u8; RESPONSE_INITIAL_BYTES_LENGTH];
stream.read(&mut response_initial_buffer).await?;
diff --git a/core/server-ng/src/auth.rs b/core/server-ng/src/auth.rs
index c02b2ca23..7e6c4627b 100644
--- a/core/server-ng/src/auth.rs
+++ b/core/server-ng/src/auth.rs
@@ -26,12 +26,12 @@ use crate::dispatch::submit_register_on_owner;
use crate::login_register::LoginRegisterError;
use crate::responses::{build_empty_reply, build_login_register_reply, current_metadata_commit};
use crate::session_manager::{ClientSdkInfo, SessionManager};
-use consensus::MetadataHandle;
+use consensus::{MetadataHandle, build_result_rejection_reply};
use iggy_binary_protocol::{ClientVersionInfo, RequestHeader};
use iggy_common::defaults::{
MAX_PASSWORD_LENGTH, MAX_USERNAME_LENGTH, MIN_PASSWORD_LENGTH, MIN_USERNAME_LENGTH,
};
-use iggy_common::{IggyTimestamp, PersonalAccessToken, UserStatus};
+use iggy_common::{IggyError, IggyTimestamp, PersonalAccessToken, UserStatus};
use message_bus::MessageBus;
use metadata::impls::metadata::StreamsFrontend;
use server_common::crypto;
@@ -210,6 +210,42 @@ pub(crate) async fn surface_login_failure(
) {
if error.is_terminal() {
send_login_failure_reply(shard, transport_client_id, request_header).await;
+ } else {
+ // Transient consensus failure (not-caught-up / not-primary / pipeline
+ // full): send the explicit `TransientNotCommitted` frame instead of
+ // staying silent, so the SDK replays the login immediately rather than
+ // waiting out its read-timeout. Same contract as a transient metadata
+ // request -- nothing committed, so the replayed Register is idempotent.
+ send_login_transient_reply(shard, transport_client_id, request_header).await;
+ }
+}
+
+/// Result-framed `TransientNotCommitted` Reply on a transient (non-terminal)
+/// failed Register. The SDK decodes the nonzero result code and replays the
+/// same login on the same connection. Only call for transient errors -- see
+/// [`surface_login_failure`].
+#[allow(clippy::future_not_send)]
+async fn send_login_transient_reply(
+ shard: &Rc<ServerNgShard>,
+ transport_client_id: u128,
+ request_header: &RequestHeader,
+) {
+ let commit = current_metadata_commit(shard);
+ let reply = build_result_rejection_reply(
+ request_header,
+ commit,
+ IggyError::TransientNotCommitted.as_code(),
+ );
+ if let Err(error) = shard
+ .bus
+ .send_to_client(transport_client_id, reply.into_generic().into_frozen())
+ .await
+ {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "failed to send login transient reply"
+ );
}
}
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index ab82265cc..46bd45c29 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -24,6 +24,7 @@ use crate::dispatch::{
use crate::partition_helpers::{
configure_consumer_offsets, ensure_initial_segment, validate_namespace_bounds,
};
+use crate::segment_recovery::{RecoveredSegment, load_persisted_segments};
use crate::server_error::{ServerNgError, ShardJoinFailure, ShardJoinFailureKind};
use crate::session_manager::SessionManager;
use configs::server_ng::ServerNgConfig;
@@ -36,7 +37,7 @@ use consensus::{LocalPipeline, MetadataHandle, PartitionsHandle, Sequencer, VsrC
// non-blocking variants for cancel-safe shutdown polling.
use crossfire::{AsyncRxTrait, AsyncTxTrait};
use iggy_binary_protocol::Operation;
-use iggy_common::{IggyByteSize, PartitionStats, TopicStats, variadic};
+use iggy_common::{IggyByteSize, PartitionStats, variadic};
use journal::Journal;
use journal::prepare_journal::PrepareJournal;
use message_bus::client_listener::{self, RequestHandler};
@@ -53,8 +54,8 @@ use message_bus::transports::tls::{
};
use message_bus::{
AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedTlsClientFn,
- AcceptedWsClientFn, DialedReplicaFn, IggyMessageBus, MAX_INFLIGHT_REPLICA_HANDSHAKES,
- ReplicaOwnerTable, connector,
+ AcceptedWsClientFn, AcceptedWssClientFn, DialedReplicaFn, IggyMessageBus,
+ MAX_INFLIGHT_REPLICA_HANDSHAKES, ReplicaOwnerTable, connector,
};
use metadata::IggyMetadata;
use metadata::MuxStateMachine;
@@ -65,7 +66,7 @@ use metadata::stm::snapshot::Snapshot;
use metadata::stm::stream::{Partition, Streams};
use metadata::stm::user::Users;
use partitions::{
- IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
+ IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig,
};
use rustls::pki_types::ServerName;
use server_common::bootstrap::create_directories;
@@ -401,6 +402,7 @@ struct LocalClientAcceptFns {
ws: AcceptedWsClientFn,
quic: AcceptedQuicClientFn,
tcp_tls: AcceptedTlsClientFn,
+ wss: AcceptedWssClientFn,
}
#[derive(Default)]
@@ -849,6 +851,9 @@ async fn shard_main(
mux_stm,
Some(PathBuf::from(&config.system.path)),
);
+ // Shard 0's copy resolves the `MaxTopicSize::ServerDefault` sentinel at
+ // admission; every shard's copy backs the same resolution in responses.
+ metadata.set_default_max_topic_size(config.system.topic.max_size.as_bytes_u64());
let shard_metrics = ShardMetrics::for_shard();
// Notifier install deferred until after tick handler wires below.
@@ -1339,7 +1344,15 @@ async fn build_shard_for_thread(
let owning_shard =
calculate_shard_assignment(&namespace, u32::from(total_shards));
if owning_shard == shard_id {
- owned.push((stream.id, topic_id, topic.stats.clone(), partition.clone()));
+ // Shared per-partition stats from the registry: the
+ // same `Arc` backs every shard's `get_topic` reply.
+ let stats = inner.stats_registry.partition(
+ stream.id,
+ topic_id,
+ partition.id,
+ topic.stats.clone(),
+ );
+ owned.push((stream.id, topic_id, stats, partition.clone()));
} else {
shards_table.insert(
namespace,
@@ -1359,13 +1372,13 @@ async fn build_shard_for_thread(
// bundle was broadcast (see `MetadataHandoff::Owner`). All shards
// here only add their per-partition deltas, so the shared
// `Arc<TopicStats>` atomics race only against other atomic adds.
- for (stream_id, topic_id, topic_stats, partition_metadata) in owned {
+ for (stream_id, topic_id, partition_stats, partition_metadata) in owned {
validate_namespace_bounds(config, stream_id, topic_id, partition_metadata.id)?;
let namespace = IggyNamespace::new(stream_id, topic_id, partition_metadata.id);
let partition = load_partition(
config,
namespace,
- topic_stats,
+ partition_stats,
&partition_metadata,
topology.cluster_id,
topology.self_replica_id,
@@ -1484,7 +1497,7 @@ fn restore_metadata_consensus(
async fn load_partition(
config: &ServerNgConfig,
namespace: IggyNamespace,
- topic_stats: Arc<TopicStats>,
+ stats: Arc<PartitionStats>,
partition_metadata: &Partition,
cluster_id: u128,
self_replica_id: u8,
@@ -1494,7 +1507,6 @@ async fn load_partition(
let stream_id = namespace.stream_id();
let topic_id = namespace.topic_id();
let partition_id = namespace.partition_id();
- let stats = Arc::new(PartitionStats::new(topic_stats));
let consensus = VsrConsensus::new(
cluster_id,
self_replica_id,
@@ -1505,29 +1517,19 @@ async fn load_partition(
);
consensus.init();
- // TODO: decouple the loading logic from the `server` crate and load directly
- // into the new `partitions` log/runtime types.
- let loaded_log = server::bootstrap::load_segments(
- &config.system,
- stream_id,
- topic_id,
- partition_id,
- config
- .system
- .get_partition_path(stream_id, topic_id, partition_id),
- stats.clone(),
- )
- .await
- .map_err(|source| {
- error!(
- stream_id,
- topic_id,
- partition_id,
- error = %source,
- "failed to load partition log during server-ng bootstrap"
- );
- source
- })?;
+ let recovered_segments =
+ load_persisted_segments(config, stream_id, topic_id, partition_id, &stats)
+ .await
+ .map_err(|source| {
+ error!(
+ stream_id,
+ topic_id,
+ partition_id,
+ error = %source,
+ "failed to load partition log during server-ng bootstrap"
+ );
+ source
+ })?;
let mut partition = IggyPartition::new(stats.clone(), consensus);
partition.set_partition_dir(config.system.get_partition_path(
@@ -1541,7 +1543,7 @@ async fn load_partition(
stream_id,
topic_id,
partition_id,
- loaded_log,
+ recovered_segments,
)
.await?;
@@ -1577,43 +1579,12 @@ async fn hydrate_partition_log(
stream_id: usize,
topic_id: usize,
partition_id: usize,
- loaded_log: server::streaming::partitions::log::SegmentedLog<
- server::streaming::partitions::journal::MemoryMessageJournal,
- >,
+ recovered_segments: Vec<RecoveredSegment>,
) -> Result<(), ServerNgError> {
- // TODO: decouple the loading logic from the `server` crate. This currently
- // adapts the old server segmented log into the new `partitions` log.
- for (segment_index, (segment, storage)) in loaded_log
- .segments()
- .iter()
- .zip(loaded_log.storages().iter().cloned())
- .enumerate()
- {
- validate_recovered_segment(
- stream_id,
- topic_id,
- partition_id,
- segment,
- &storage,
- loaded_log
- .indexes()
- .get(segment_index)
- .and_then(|indexes| indexes.as_ref()),
- )?;
- let max_timestamp = match loaded_log
- .indexes()
- .get(segment_index)
- .and_then(|indexes| indexes.as_ref())
- {
- Some(indexes) => indexes_max_timestamp(indexes),
- None => load_segment_max_timestamp(&storage, stream_id, topic_id, partition_id).await?,
- };
- partition.log.add_persisted_segment(
- convert_segment(segment, max_timestamp),
- storage,
- None,
- None,
- );
+ for RecoveredSegment { segment, storage } in recovered_segments {
+ partition
+ .log
+ .add_persisted_segment(segment, storage, None, None);
}
if let Some(active_index) = partition.log.segments().len().checked_sub(1) {
@@ -1670,85 +1641,6 @@ async fn hydrate_partition_log(
Ok(())
}
-fn validate_recovered_segment(
- stream_id: usize,
- topic_id: usize,
- partition_id: usize,
- segment: &iggy_common::Segment,
- storage: &server_common::SegmentStorage,
- indexes: Option<&server::streaming::segments::IggyIndexesMut>,
-) -> Result<(), ServerNgError> {
- let messages_size_bytes = storage
- .messages_reader
- .as_ref()
- .map_or(0, |reader| u64::from(reader.file_size()));
- let indexed_size_bytes = indexes.map_or(0, |indexes| u64::from(indexes.messages_size()));
- if messages_size_bytes == indexed_size_bytes {
- return Ok(());
- }
-
- Err(ServerNgError::RecoveredSegmentSizeDivergence {
- stream_id,
- topic_id,
- partition_id,
- start_offset: segment.start_offset,
- end_offset: segment.end_offset,
- messages_size_bytes,
- indexed_size_bytes,
- })
-}
-
-fn convert_segment(segment: &iggy_common::Segment, max_timestamp: u64) -> Segment {
- Segment {
- sealed: segment.sealed,
- start_timestamp: segment.start_timestamp,
- end_timestamp: segment.end_timestamp,
- max_timestamp,
- current_position: u64::from(segment.current_position),
- start_offset: segment.start_offset,
- end_offset: segment.end_offset,
- size: segment.size,
- max_size: segment.max_size,
- }
-}
-
-fn indexes_max_timestamp(indexes: &server::streaming::segments::IggyIndexesMut) -> u64 {
- let mut max_timestamp = 0;
- for index in 0..indexes.count() {
- if let Some(index_view) = indexes.get(index) {
- max_timestamp = max_timestamp.max(index_view.timestamp());
- }
- }
-
- max_timestamp
-}
-
-async fn load_segment_max_timestamp(
- storage: &server_common::SegmentStorage,
- stream_id: usize,
- topic_id: usize,
- partition_id: usize,
-) -> Result<u64, ServerNgError> {
- let Some(index_reader) = storage.index_reader.as_ref() else {
- return Ok(0);
- };
-
- let indexes = index_reader
- .load_all_indexes_from_disk()
- .await
- .map_err(|source| {
- error!(
- stream_id,
- topic_id,
- partition_id,
- error = %source,
- "failed to load segment indexes while recovering max timestamp"
- );
- source
- })?;
- Ok(indexes_max_timestamp(&indexes))
-}
-
fn resolve_tcp_topology(
config: &ServerNgConfig,
current_replica_id: Option<u8>,
@@ -1941,6 +1833,9 @@ async fn start_tcp_runtime(
.await
}
+// ws/wss bindings intentionally mirror the transport names (same convention as
+// `replica_io::start_on_shard_zero`).
+#[allow(clippy::similar_names)]
async fn start_via_replica_io(
shard: &Rc<ServerNgShard>,
config: &ServerNgConfig,
@@ -1962,34 +1857,46 @@ async fn start_via_replica_io(
.is_some()
.then(|| load_tcp_tls_server_credentials(config))
.transpose()?;
+ // `websocket.tls.enabled` upgrades the websocket address to a WSS
+ // listener; the plain-WS listener must NOT also bind it (one port, one
+ // handshake kind -- a plain upgrade parser fed a TLS ClientHello rejects
+ // every connection with an httparse error).
+ let wss_enabled = config.websocket.tls.enabled;
+ let ws_listen_addr = (!wss_enabled).then_some(topology.ws_listen_addr).flatten();
+ let wss_listen_addr = wss_enabled.then_some(topology.ws_listen_addr).flatten();
+ let wss_credentials = wss_listen_addr
+ .is_some()
+ .then(|| load_wss_server_credentials(config))
+ .transpose()?;
let LocalClientAcceptFns {
tcp,
ws,
quic,
tcp_tls,
+ wss,
} = accepted_clients;
let bound = replica_io::start_on_shard_zero(
&shard.bus,
replica_addr,
topology.client_listen_addr,
- topology.ws_listen_addr,
+ ws_listen_addr,
topology.quic_listen_addr,
quic_credentials,
topology.tcp_tls_listen_addr,
tcp_tls_credentials,
- None,
- None,
+ wss_listen_addr,
+ wss_credentials,
topology.self_replica_id,
topology.peers.clone(),
accepted_replica,
dialed_replica,
tcp,
- topology.ws_listen_addr.map(|_| ws),
+ ws_listen_addr.map(|_| ws),
topology.quic_listen_addr.map(|_| quic),
topology.tcp_tls_listen_addr.map(|_| tcp_tls),
- None,
+ wss_listen_addr.map(|_| wss),
shard.bus.config().reconnect_period,
)
.await
@@ -2013,7 +1920,8 @@ async fn start_via_replica_io(
config.cluster.enabled.then_some(bound.replica),
bound.tcp_tls,
bound.quic,
- bound.ws,
+ // The WSS listener occupies the configured websocket address slot.
+ bound.wss.or(bound.ws),
)
.await?;
if config.cluster.enabled {
@@ -2213,6 +2121,9 @@ fn make_replica_delegation_fns(
/// locally on shard 0 because their per-connection state is not portable
/// across shards (`compio_quic` endpoint binds one UDP socket; rustls TLS
/// state ties to the post-handshake reactor).
+// ws/wss bindings intentionally mirror the transport names (same convention as
+// `replica_io::start_on_shard_zero`).
+#[allow(clippy::similar_names)]
fn make_shard_zero_client_accept_fns(
coord: Rc<shard::coordinator::ShardZeroCoordinator>,
bus: &Rc<IggyMessageBus>,
@@ -2220,7 +2131,9 @@ fn make_shard_zero_client_accept_fns(
) -> LocalClientAcceptFns {
let quic_bus = Rc::clone(bus);
let tcp_tls_bus = Rc::clone(bus);
+ let wss_bus = Rc::clone(bus);
let quic_request = on_request.clone();
+ let wss_request = on_request.clone();
let tcp_tls_request = on_request;
let tcp_coord = Rc::clone(&coord);
@@ -2247,7 +2160,7 @@ fn make_shard_zero_client_accept_fns(
installer::install_client_quic(&quic_bus, meta, accepted, quic_request.clone());
});
- let tcp_tls_coord = coord;
+ let tcp_tls_coord = Rc::clone(&coord);
let tcp_tls = Rc::new(move |stream, tls_config| {
let Some(meta) =
client_meta_from_stream(&stream, &tcp_tls_coord, ClientTransportKind::TcpTls)
@@ -2263,11 +2176,24 @@ fn make_shard_zero_client_accept_fns(
);
});
+ // WSS terminates locally on shard 0 like TCP-TLS (rustls state is not
+ // serialisable across the delegate path), minting ids through the same
+ // coordinator counter.
+ let wss_coord = coord;
+ let wss = Rc::new(move |stream, tls_config| {
+ let Some(meta) = client_meta_from_stream(&stream, &wss_coord, ClientTransportKind::Wss)
+ else {
+ return;
+ };
+ installer::install_client_wss(&wss_bus, meta, stream, tls_config, wss_request.clone());
+ });
+
LocalClientAcceptFns {
tcp,
ws,
quic,
tcp_tls,
+ wss,
}
}
@@ -2323,18 +2249,7 @@ async fn start_client_listeners(
}
if let Some(ws_addr) = topology.ws_listen_addr {
- let (listener, bound_addr) =
- client_listener::ws::bind(ws_addr).await.map_err(|source| {
- error!(addr = %ws_addr, error = %source, "failed to bind websocket listener");
- source
- })?;
- let token = shard.bus.token();
- let accepted_ws = accepted_clients.ws.clone();
- let ws_handle = compio::runtime::spawn(async move {
- client_listener::ws::run(listener, token, accepted_ws).await;
- });
- shard.bus.track_background(ws_handle);
- bound.ws = Some(bound_addr);
+ bound.ws = Some(start_websocket_listener(shard, config, ws_addr, accepted_clients).await?);
}
if let Some(quic_addr) = topology.quic_listen_addr {
@@ -2527,6 +2442,62 @@ fn load_tcp_tls_server_credentials(
})
}
+/// Bind the websocket client listener on `ws_addr`: WSS when
+/// `websocket.tls.enabled` (the plain-WS accept loop must not also bind the
+/// port -- a plain upgrade parser fed a TLS `ClientHello` rejects every
+/// connection with an httparse error), plain WS otherwise.
+async fn start_websocket_listener(
+ shard: &Rc<ServerNgShard>,
+ config: &ServerNgConfig,
+ ws_addr: SocketAddr,
+ accepted_clients: &LocalClientAcceptFns,
+) -> Result<SocketAddr, ServerNgError> {
+ if config.websocket.tls.enabled {
+ let credentials = load_wss_server_credentials(config)?;
+ let (listener, tls_config, bound_addr) = client_listener::wss::bind(ws_addr, credentials)
+ .map_err(|source| {
+ error!(addr = %ws_addr, error = %source, "failed to bind WSS listener");
+ source
+ })?;
+ let token = shard.bus.token();
+ let accepted_wss = accepted_clients.wss.clone();
+ let wss_handle = compio::runtime::spawn(async move {
+ client_listener::wss::run(listener, tls_config, token, accepted_wss).await;
+ });
+ shard.bus.track_background(wss_handle);
+ Ok(bound_addr)
+ } else {
+ let (listener, bound_addr) =
+ client_listener::ws::bind(ws_addr).await.map_err(|source| {
+ error!(addr = %ws_addr, error = %source, "failed to bind websocket listener");
+ source
+ })?;
+ let token = shard.bus.token();
+ let accepted_ws = accepted_clients.ws.clone();
+ let ws_handle = compio::runtime::spawn(async move {
+ client_listener::ws::run(listener, token, accepted_ws).await;
+ });
+ shard.bus.track_background(ws_handle);
+ Ok(bound_addr)
+ }
+}
+
+fn load_wss_server_credentials(
+ config: &ServerNgConfig,
+) -> Result<TlsServerCredentials, ServerNgError> {
+ let tls = &config.websocket.tls;
+ if tls.self_signed && !Path::new(&tls.cert_file).exists() {
+ return Ok(self_signed_for_loopback());
+ }
+
+ load_pem(Path::new(&tls.cert_file), Path::new(&tls.key_file)).map_err(|source| {
+ ServerNgError::ListenerCredentials {
+ transport: "websocket.tls",
+ source,
+ }
+ })
+}
+
fn load_quic_server_credentials(
config: &ServerNgConfig,
) -> Result<replica_io::QuicServerCredentials, ServerNgError> {
diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs
index 74b81e92b..421ebf25c 100644
--- a/core/server-ng/src/dispatch.rs
+++ b/core/server-ng/src/dispatch.rs
@@ -34,9 +34,9 @@ use crate::login_register::LoginRegisterError;
use crate::pat::maybe_rewrite_pat_request;
use crate::responses::{
NonReplicatedResponse, build_consumer_offset_body, build_empty_reply, build_get_me_response,
- build_non_replicated_response, build_polled_messages_body, build_raw_pat_reply,
- connected_client_to_response, current_metadata_commit, resolve_partition_namespace,
- resolve_partition_request_namespace,
+ build_get_personal_access_tokens_response, build_non_replicated_response,
+ build_polled_messages_body, build_raw_pat_reply, connected_client_to_response,
+ current_metadata_commit, resolve_partition_namespace, resolve_partition_request_namespace,
};
use crate::session_manager::SessionManager;
use crate::users::maybe_rewrite_user_password_request;
@@ -48,7 +48,8 @@ use consensus::{
};
use iggy_binary_protocol::codes::{
GET_CLIENT_CODE, GET_CLIENTS_CODE, GET_CLUSTER_METADATA_CODE, GET_CONSUMER_OFFSET_CODE,
- GET_ME_CODE, PING_CODE, POLL_MESSAGES_CODE, SYNC_CONSUMER_GROUP_CODE,
+ GET_ME_CODE, GET_PERSONAL_ACCESS_TOKENS_CODE, PING_CODE, POLL_MESSAGES_CODE,
+ SYNC_CONSUMER_GROUP_CODE,
};
use iggy_binary_protocol::primitives::consumer::WireConsumer;
use iggy_binary_protocol::primitives::polling_strategy::WirePollingStrategy;
@@ -732,6 +733,48 @@ async fn handle_client_request(
}
}
+/// Per-user PATs, resolved from this shard's session (like `get_me`) and read
+/// out of the Users STM. Built here rather than in `build_non_replicated_response`
+/// which has no session context.
+#[allow(clippy::future_not_send)]
+async fn handle_get_personal_access_tokens(
+ shard: &Rc<ServerNgShard>,
+ sessions: &Rc<RefCell<SessionManager>>,
+ transport_client_id: u128,
+ request: &Message<RequestHeader>,
+) {
+ let response = build_get_personal_access_tokens_response(shard, sessions, transport_client_id);
+ send_non_replicated_bytes(
+ shard,
+ request,
+ transport_client_id,
+ response.to_bytes(),
+ "get_personal_access_tokens",
+ )
+ .await;
+}
+
+/// The requesting connection's own identity, sourced from this shard's
+/// `SessionManager` (not `IggyMetadata`), so built here rather than in
+/// `build_non_replicated_response`.
+#[allow(clippy::future_not_send)]
+async fn handle_get_me(
+ shard: &Rc<ServerNgShard>,
+ sessions: &Rc<RefCell<SessionManager>>,
+ transport_client_id: u128,
+ request: &Message<RequestHeader>,
+) {
+ let response = build_get_me_response(shard, sessions, transport_client_id);
+ send_non_replicated_bytes(
+ shard,
+ request,
+ transport_client_id,
+ response.to_bytes(),
+ "get_me",
+ )
+ .await;
+}
+
#[allow(clippy::future_not_send)]
async fn handle_non_replicated_request(
shard: &Rc<ServerNgShard>,
@@ -766,19 +809,10 @@ async fn handle_non_replicated_request(
}
}
GET_ME_CODE => {
- // `get_me` reports the requesting connection's own identity,
- // sourced from this shard's `SessionManager` (not `IggyMetadata`),
- // so it is built here rather than in
- // `build_non_replicated_response`.
- let response = build_get_me_response(shard, sessions, transport_client_id);
- send_non_replicated_bytes(
- shard,
- &request,
- transport_client_id,
- response.to_bytes(),
- "get_me",
- )
- .await;
+ handle_get_me(shard, sessions, transport_client_id, &request).await;
+ }
+ GET_PERSONAL_ACCESS_TOKENS_CODE => {
+ handle_get_personal_access_tokens(shard, sessions, transport_client_id, &request).await;
}
GET_CLIENTS_CODE => {
// Shared-nothing: each shard knows only its own connections, so
diff --git a/core/server-ng/src/lib.rs b/core/server-ng/src/lib.rs
index 78d9847b8..22ead12fa 100644
--- a/core/server-ng/src/lib.rs
+++ b/core/server-ng/src/lib.rs
@@ -29,6 +29,7 @@ pub mod pat;
pub(crate) mod personal_access_token_cleaner;
pub mod responses;
pub(crate) mod segment_cleaner;
+pub(crate) mod segment_recovery;
pub mod server_error;
pub mod session_manager;
pub mod users;
diff --git a/core/server-ng/src/partition_helpers.rs b/core/server-ng/src/partition_helpers.rs
index 21ba99db9..6653819ad 100644
--- a/core/server-ng/src/partition_helpers.rs
+++ b/core/server-ng/src/partition_helpers.rs
@@ -29,7 +29,7 @@ use compio::fs::create_dir_all;
use configs::server_ng::ServerNgConfig;
use consensus::{LocalPipeline, VsrConsensus};
use iggy_common::{
- ConsumerGroupOffsets, ConsumerOffsets, IggyError, IggyTimestamp, PartitionStats, TopicStats,
+ ConsumerGroupOffsets, ConsumerOffsets, IggyError, IggyTimestamp, PartitionStats,
};
use message_bus::IggyMessageBus;
use partitions::{IggyIndexWriter, IggyPartition, MessagesWriter, Segment};
@@ -428,7 +428,7 @@ pub async fn ensure_initial_segment(
pub async fn build_partition_fresh(
config: &ServerNgConfig,
namespace: IggyNamespace,
- topic_stats: Arc<TopicStats>,
+ stats: Arc<PartitionStats>,
cluster_id: u128,
self_replica_id: u8,
replica_count: u8,
@@ -452,7 +452,6 @@ pub async fn build_partition_fresh(
source
})?;
- let stats = Arc::new(PartitionStats::new(topic_stats));
let consensus = VsrConsensus::new(
cluster_id,
self_replica_id,
diff --git a/core/server-ng/src/partition_reconciler.rs b/core/server-ng/src/partition_reconciler.rs
index 62cad7ea4..ca0da51cd 100644
--- a/core/server-ng/src/partition_reconciler.rs
+++ b/core/server-ng/src/partition_reconciler.rs
@@ -420,18 +420,18 @@ async fn reconcile_additions(
continue;
}
- // Clone the parent `Arc<TopicStats>` only for namespaces actually
+ // Resolve the shared stats `Arc` only for namespaces actually
// built, not once per committed partition every pass. A topic that
// vanished between the target snapshot and this read defers to the
// next pass.
- let Some(topic_stats) = fetch_topic_stats(ctx, ns) else {
+ let Some(partition_stats) = fetch_partition_stats(ctx, ns) else {
continue;
};
match build_partition_fresh(
ctx.config.as_ref(),
ns,
- topic_stats,
+ partition_stats,
ctx.cluster_id,
ctx.self_replica_id,
ctx.replica_count,
@@ -712,7 +712,7 @@ fn snapshot_topic_live_groups(ctx: &ReconcilerCtx) -> AHashMap<(usize, usize), A
/// Committed `(namespace, created_revision)` pairs. The epoch lets the
/// additions pass detect a stale local incarnation after slab-key reuse
/// without an `Arc<TopicStats>` clone per partition; stats are fetched
-/// lazily in [`fetch_topic_stats`] only for namespaces actually built.
+/// lazily in [`fetch_partition_stats`] only for namespaces actually built.
fn snapshot_target_namespaces(ctx: &ReconcilerCtx) -> Vec<(IggyNamespace, u64)> {
ctx.shard.plane.metadata().mux_stm.streams().read(|inner| {
// TODO(krishna): O(committed partitions) per non-skipped pass (here +
@@ -745,14 +745,21 @@ fn current_revision(ctx: &ReconcilerCtx) -> u64 {
/// Clone the parent topic's `Arc<TopicStats>` for a single namespace.
/// `None` if the topic vanished between the target snapshot and this read.
-fn fetch_topic_stats(
+fn fetch_partition_stats(
ctx: &ReconcilerCtx,
ns: IggyNamespace,
-) -> Option<Arc<iggy_common::TopicStats>> {
+) -> Option<Arc<iggy_common::PartitionStats>> {
ctx.shard.plane.metadata().mux_stm.streams().read(|inner| {
let stream = inner.items.get(ns.stream_id())?;
let topic = stream.topics.get(ns.topic_id())?;
- Some(topic.stats.clone())
+ // Get-or-create in the shared registry so the owning shard's counters
+ // are the same `Arc` every shard's `get_topic` reply reads.
+ Some(inner.stats_registry.partition(
+ ns.stream_id(),
+ ns.topic_id(),
+ ns.partition_id(),
+ topic.stats.clone(),
+ ))
})
}
diff --git a/core/server-ng/src/responses.rs b/core/server-ng/src/responses.rs
index bd11b9915..f3a0078d5 100644
--- a/core/server-ng/src/responses.rs
+++ b/core/server-ng/src/responses.rs
@@ -33,6 +33,7 @@ use iggy_binary_protocol::codes::{
GET_STREAM_CODE, GET_STREAMS_CODE, GET_TOPIC_CODE, GET_TOPICS_CODE, GET_USER_CODE,
GET_USERS_CODE,
};
+use iggy_binary_protocol::consensus::{RESULT_COUNT_LEN, result_code};
use iggy_binary_protocol::primitives::consumer::WireConsumer;
use iggy_binary_protocol::requests::consumer_groups::{
GetConsumerGroupRequest, GetConsumerGroupsRequest,
@@ -51,6 +52,9 @@ use iggy_binary_protocol::responses::clients::client_response::ConsumerGroupInfo
use iggy_binary_protocol::responses::clients::get_client::ClientDetailsResponse;
use iggy_binary_protocol::responses::consumer_groups::GetConsumerGroupsResponse;
use iggy_binary_protocol::responses::personal_access_tokens::RawPersonalAccessTokenResponse;
+use iggy_binary_protocol::responses::personal_access_tokens::get_personal_access_tokens::{
+ GetPersonalAccessTokensResponse, PersonalAccessTokenResponse,
+};
use iggy_binary_protocol::responses::streams::StreamResponse;
use iggy_binary_protocol::responses::streams::get_stream::{
GetStreamResponse, TopicHeader as StreamTopicHeader,
@@ -84,6 +88,36 @@ use std::rc::Rc;
/// (`user_id`, transport kind, peer address) comes from the per-shard
/// [`SessionManager`]; the `consumer_groups` list is read from the
/// (replicated) consumer-group STM by the connection's bound VSR client id.
+pub(crate) fn build_get_personal_access_tokens_response(
+ shard: &Rc<ServerNgShard>,
+ sessions: &Rc<RefCell<SessionManager>>,
+ transport_client_id: u128,
+) -> GetPersonalAccessTokensResponse {
+ // PATs are per-user; list the requesting connection's own tokens, resolved
+ // from this shard's `SessionManager` (like `get_me`) then read out of the
+ // replicated Users STM.
+ let Some(user_id) = sessions.borrow().get_user_id(transport_client_id) else {
+ return GetPersonalAccessTokensResponse { tokens: Vec::new() };
+ };
+ shard.plane.metadata().mux_stm.users().read(|users| {
+ let tokens = users
+ .personal_access_tokens
+ .get(&user_id)
+ .map(|pats| {
+ pats.values()
+ .filter_map(|pat| {
+ Some(PersonalAccessTokenResponse {
+ name: WireName::new(pat.name.as_ref()).ok()?,
+ expiry_at: pat.expiry_at.map_or(0, |expiry| expiry.as_micros()),
+ })
+ })
+ .collect()
+ })
+ .unwrap_or_default();
+ GetPersonalAccessTokensResponse { tokens }
+ })
+}
+
pub(crate) fn build_get_me_response(
shard: &Rc<ServerNgShard>,
sessions: &Rc<RefCell<SessionManager>>,
@@ -476,36 +510,48 @@ fn build_cluster_metadata_response() -> ClusterMetadataResponse {
}
fn build_stats_response(shard: &Rc<ServerNgShard>) -> Result<StatsResponse, IggyError> {
- let (streams_count, topics_count, partitions_count, messages_size_bytes, messages_count) =
- shard
- .plane
- .metadata()
- .mux_stm
- .streams()
- .read(|streams| -> Result<_, IggyError> {
- let mut topics_count = 0u32;
- let mut partitions_count = 0u32;
- let mut messages_size_bytes = 0u64;
- let mut messages_count = 0u64;
- for (_, stream) in &streams.items {
- topics_count = topics_count.saturating_add(usize_to_u32(stream.topics.len())?);
- messages_size_bytes =
- messages_size_bytes.saturating_add(stream.stats.size_bytes_inconsistent());
- messages_count =
- messages_count.saturating_add(stream.stats.messages_count_inconsistent());
- for (_, topic) in &stream.topics {
- partitions_count =
- partitions_count.saturating_add(usize_to_u32(topic.partitions.len())?);
- }
+ let (
+ streams_count,
+ topics_count,
+ partitions_count,
+ segments_count,
+ messages_size_bytes,
+ messages_count,
+ ) = shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .read(|streams| -> Result<_, IggyError> {
+ let mut topics_count = 0u32;
+ let mut partitions_count = 0u32;
+ let mut segments_count = 0u32;
+ let mut messages_size_bytes = 0u64;
+ let mut messages_count = 0u64;
+ for (_, stream) in &streams.items {
+ topics_count = topics_count.saturating_add(usize_to_u32(stream.topics.len())?);
+ messages_size_bytes =
+ messages_size_bytes.saturating_add(stream.stats.size_bytes_inconsistent());
+ messages_count =
+ messages_count.saturating_add(stream.stats.messages_count_inconsistent());
+ // Segment counts roll up from the partition plane through
+ // the shared stats registry (partition -> topic -> stream).
+ segments_count =
+ segments_count.saturating_add(stream.stats.segments_count_inconsistent());
+ for (_, topic) in &stream.topics {
+ partitions_count =
+ partitions_count.saturating_add(usize_to_u32(topic.partitions.len())?);
}
- Ok((
- usize_to_u32(streams.items.len())?,
- topics_count,
- partitions_count,
- messages_size_bytes,
- messages_count,
- ))
- })?;
+ }
+ Ok((
+ usize_to_u32(streams.items.len())?,
+ topics_count,
+ partitions_count,
+ segments_count,
+ messages_size_bytes,
+ messages_count,
+ ))
+ })?;
let consumer_groups_count = usize_to_u32(
shard
.plane
@@ -530,7 +576,7 @@ fn build_stats_response(shard: &Rc<ServerNgShard>) -> Result<StatsResponse, Iggy
streams_count,
topics_count,
partitions_count,
- segments_count: 0,
+ segments_count,
messages_count,
clients_count: 0,
consumer_groups_count,
@@ -653,7 +699,7 @@ fn build_get_topic_response(
partitions: topic
.partitions
.iter()
- .map(partition_response)
+ .map(|partition| partition_response(streams, stream_id, topic_id, partition))
.collect::<Result<Vec<_>, _>>()?,
}))
})
@@ -735,15 +781,34 @@ fn topic_header(topic: &metadata::stm::stream::Topic) -> Result<StreamTopicHeade
}
fn partition_response(
+ streams: &metadata::stm::stream::StreamsInner,
+ stream_id: usize,
+ topic_id: usize,
partition: &metadata::stm::stream::Partition,
) -> Result<PartitionResponse, IggyError> {
+ // Per-partition counters live in the shared stats registry (one `Arc`
+ // across all shards and both left-right buffers), populated when the
+ // owning shard materializes the partition; `None` only in the window
+ // before that first materialization.
+ let stats = streams
+ .stats_registry
+ .partition_get(stream_id, topic_id, partition.id);
+ let (segments_count, current_offset, size_bytes, messages_count) =
+ stats.map_or((0, 0, 0, 0), |stats| {
+ (
+ stats.segments_count_inconsistent(),
+ stats.current_offset(),
+ stats.size_bytes_inconsistent(),
+ stats.messages_count_inconsistent(),
+ )
+ });
Ok(PartitionResponse {
id: usize_to_u32(partition.id)?,
created_at: partition.created_at.as_micros(),
- segments_count: 0,
- current_offset: 0,
- size_bytes: 0,
- messages_count: 0,
+ segments_count,
+ current_offset,
+ size_bytes,
+ messages_count,
})
}
@@ -788,14 +853,28 @@ pub(crate) fn build_login_register_reply(
commit: u64,
user_id: u32,
) -> Message<ReplyHeader> {
- let body = LoginRegisterResponse {
+ // Result-framed like every metadata reply: a zero result-count (success)
+ // followed by the `LoginRegisterResponse` payload. A transient Register
+ // instead ships a `[count=1][index=0][TransientNotCommitted]` frame
+ // (`build_transient_reply`), which the SDK decodes and replays. The matching
+ // strip is in the SDK `split_metadata_result` (Register is result-framed).
+ let payload = LoginRegisterResponse {
user_id,
session,
server_protocol_version: IGGY_PROTOCOL_VERSION,
server_version: WireName::new(SERVER_VERSION).expect("SERVER_VERSION is 1-255 bytes"),
}
.to_bytes();
- build_reply_from_bytes(request_header, client_id, session, commit, &body)
+ let mut body = Vec::with_capacity(RESULT_COUNT_LEN + payload.len());
+ body.extend_from_slice(&[0u8; RESULT_COUNT_LEN]);
+ body.extend_from_slice(&payload);
+ build_reply_from_bytes(
+ request_header,
+ client_id,
+ session,
+ commit,
+ &Bytes::from(body),
+ )
}
pub(crate) fn build_reply_from_bytes(
@@ -839,12 +918,30 @@ pub(crate) fn build_raw_pat_reply(
return Ok(committed);
}
let header_len = std::mem::size_of::<ReplyHeader>();
+ // A `Reply` whose result section is nonzero is not a successful commit but a
+ // committed business rejection or a `TransientNotCommitted` retry frame, both
+ // with no payload and no token to ship. Pass it through so the client decodes
+ // the typed result (and, for a transient, replays) instead of having a raw
+ // token grafted onto a rejection body.
+ if result_code(&committed.as_slice()[header_len..]) != Some(0) {
+ return Ok(committed);
+ }
let committed_header =
bytemuck::checked::try_from_bytes::<ReplyHeader>(&committed.as_slice()[..header_len])
.map_err(|_| IggyError::InvalidFormat)?;
let commit = committed_header.commit;
let token = WireName::new(raw.as_str()).map_err(|_| IggyError::InvalidFormat)?;
- let body = RawPersonalAccessTokenResponse { token }.to_bytes();
+ let token_payload = RawPersonalAccessTokenResponse { token }.to_bytes();
+ // Metadata reply bodies are framed `[result_count:u32][payload]`; a success
+ // is `count == 0` followed by the payload (see `ApplyReply::write_reply_body`
+ // / `result_code`). The committed reply carried an empty payload with this
+ // same framing, so reproduce it here - prepend the zero result-count rather
+ // than shipping a bare `RawPersonalAccessTokenResponse`, which the client
+ // would misread as the result-count and reject as a bogus error.
+ let mut framed = Vec::with_capacity(RESULT_COUNT_LEN + token_payload.len());
+ framed.extend_from_slice(&[0u8; RESULT_COUNT_LEN]);
+ framed.extend_from_slice(&token_payload);
+ let body = Bytes::from(framed);
let reply = build_reply_from_bytes(
request_header,
request_header.client,
diff --git a/core/server-ng/src/segment_recovery.rs b/core/server-ng/src/segment_recovery.rs
new file mode 100644
index 000000000..d205208b1
--- /dev/null
+++ b/core/server-ng/src/segment_recovery.rs
@@ -0,0 +1,306 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! server-ng-owned segment recovery.
+//!
+//! Previously the bootstrap path borrowed `server::bootstrap::load_segments`
+//! from the legacy `server` crate to hydrate persisted segments. That loader
+//! reads the legacy 16-byte dense per-message index through
+//! `server_common::IndexReader`, but server-ng persists a 24-byte sparse index
+//! (`partitions::IggyIndexWriter`: one entry per flush, absolute `offset`,
+//! `timestamp`, and batch-start `position`). Reading the 24-byte file with the
+//! 16-byte parser mis-strides it (the "Index data must be exactly 16 bytes"
+//! recovery panic). This module is the server-ng-owned loader, reading the same
+//! 24-byte format its writer emits.
+
+use crate::server_error::ServerNgError;
+use configs::server_ng::ServerNgConfig;
+use iggy_common::{IggyByteSize, IggyError, PartitionStats};
+use partitions::{IggyIndexReader, Segment};
+use server_common::SegmentStorage;
+use server_common::send_messages2::{COMMAND_HEADER_SIZE, SendMessages2Header};
+use std::fs;
+use std::os::unix::fs::FileExt;
+use tracing::{error, warn};
+
+const LOG_EXTENSION: &str = "log";
+
+/// A persisted segment recovered from disk: its metadata plus the storage
+/// handles (readers/writers) opened over its `.log` / `.index` files.
+pub struct RecoveredSegment {
+ pub segment: Segment,
+ pub storage: SegmentStorage,
+}
+
+/// Loads every persisted segment for a partition, sorted by start offset.
+///
+/// Segment offsets and timestamps are recovered from the 24-byte sparse index
+/// (see module docs); segment byte size comes from the `.log` file. The last
+/// segment is left unsealed so it can accept further writes.
+///
+/// # Errors
+///
+/// Returns an error if the partition directory or a segment's files cannot be
+/// read, or if a segment's index references a batch beyond the end of its
+/// messages file (torn write).
+pub async fn load_persisted_segments(
+ config: &ServerNgConfig,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+ stats: &PartitionStats,
+) -> Result<Vec<RecoveredSegment>, ServerNgError> {
+ let partition_path = config
+ .system
+ .get_partition_path(stream_id, topic_id, partition_id);
+ let mut start_offsets = collect_segment_start_offsets(&partition_path)?;
+ start_offsets.sort_unstable();
+
+ let enforce_fsync = config.system.partition.enforce_fsync;
+ let max_size = config.system.segment.size;
+
+ let mut recovered = Vec::with_capacity(start_offsets.len());
+ for start_offset in start_offsets {
+ let messages_path =
+ config
+ .system
+ .get_messages_file_path(stream_id, topic_id, partition_id, start_offset);
+ let index_path =
+ config
+ .system
+ .get_index_path(stream_id, topic_id, partition_id, start_offset);
+
+ let messages_size = file_len(&messages_path);
+ let index_size = file_len(&index_path);
+
+ let bounds = recover_segment_bounds(
+ &index_path,
+ &messages_path,
+ start_offset,
+ messages_size,
+ stream_id,
+ topic_id,
+ partition_id,
+ )
+ .await?;
+
+ // An index without a single whole entry means any log bytes were torn
+ // off mid-write (the crash landed between the message write and the
+ // index write). Recover the segment as EMPTY: counting the bytes with
+ // `end_offset == start_offset` would fabricate one phantom message for
+ // the bootstrap non-empty filters, and appending after the torn bytes
+ // would strand undecodable garbage inside the readable range. Zeroed
+ // sizes make the next append overwrite the torn bytes instead.
+ let (start_timestamp, end_timestamp, end_offset, effective_messages_size) =
+ if let Some((start_timestamp, end_timestamp, end_offset)) = bounds {
+ (start_timestamp, end_timestamp, end_offset, messages_size)
+ } else {
+ if messages_size > 0 {
+ warn!(
+ stream_id,
+ topic_id,
+ partition_id,
+ start_offset,
+ messages_size,
+ "segment log holds bytes but its index holds no whole \
+ entry (torn write); recovering the segment as empty"
+ );
+ }
+ (0, 0, start_offset, 0)
+ };
+ let effective_index_size = if bounds.is_some() { index_size } else { 0 };
+
+ let storage = SegmentStorage::new(
+ &messages_path,
+ &index_path,
+ effective_messages_size,
+ effective_index_size,
+ enforce_fsync,
+ enforce_fsync,
+ true,
+ )
+ .await
+ .map_err(|source| {
+ error!(
+ stream_id,
+ topic_id,
+ partition_id,
+ path = %messages_path,
+ error = %source,
+ "failed to open persisted segment storage during recovery"
+ );
+ source
+ })?;
+
+ let mut segment = Segment::new(start_offset, max_size);
+ segment.sealed = true;
+ segment.start_timestamp = start_timestamp;
+ segment.end_timestamp = end_timestamp;
+ segment.max_timestamp = end_timestamp;
+ segment.end_offset = end_offset;
+ segment.size = IggyByteSize::from(effective_messages_size);
+ segment.current_position = effective_messages_size;
+
+ stats.increment_segments_count(1);
+ stats.increment_size_bytes(effective_messages_size);
+ if effective_messages_size > 0 {
+ // Offsets in a segment are contiguous, so the message count is the
+ // inclusive span between the first (segment start) and last offset.
+ stats.increment_messages_count(end_offset - start_offset + 1);
+ }
+
+ recovered.push(RecoveredSegment { segment, storage });
+ }
+
+ if let Some(last) = recovered.last_mut() {
+ last.segment.sealed = false;
+ }
+
+ Ok(recovered)
+}
+
+/// Parses the zero-padded start offset out of every `.log` file name in the
+/// partition directory. A missing directory means a never-persisted partition.
+fn collect_segment_start_offsets(partition_path: &str) -> Result<Vec<u64>, ServerNgError> {
+ let entries = match fs::read_dir(partition_path) {
+ Ok(entries) => entries,
+ Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
+ Err(source) => {
+ error!(
+ partition_path,
+ error = %source,
+ "failed to list partition directory during recovery"
+ );
+ return Err(IggyError::CannotReadPartitions.into());
+ }
+ };
+
+ let mut start_offsets = Vec::new();
+ for entry in entries.flatten() {
+ let path = entry.path();
+ if path.extension().and_then(|ext| ext.to_str()) != Some(LOG_EXTENSION) {
+ continue;
+ }
+ if let Some(start_offset) = path
+ .file_stem()
+ .and_then(|stem| stem.to_str())
+ .and_then(|stem| stem.parse::<u64>().ok())
+ {
+ start_offsets.push(start_offset);
+ }
+ }
+
+ Ok(start_offsets)
+}
+
+fn file_len(path: &str) -> u64 {
+ fs::metadata(path).map_or(0, |metadata| metadata.len())
+}
+
+/// Derives `(start_timestamp, end_timestamp, end_offset)` from a segment's
+/// 24-byte sparse index. `None` when the index holds no whole entry (the
+/// caller recovers the segment as empty). The last entry's `position` is only
+/// the last flushed batch's START byte, so the batch header is read back from
+/// the messages file to prove the batch also ENDS inside it -- without
+/// `enforce_fsync` there is no ordering barrier between the message write and
+/// the index write, and a tail torn mid-flush would otherwise pass while
+/// `end_offset` claims offsets whose bytes are incomplete.
+async fn recover_segment_bounds(
+ index_path: &str,
+ messages_path: &str,
+ start_offset: u64,
+ messages_size: u64,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> Result<Option<(u64, u64, u64)>, ServerNgError> {
+ let reader = IggyIndexReader::new(index_path).await.map_err(|source| {
+ error!(
+ stream_id,
+ topic_id,
+ partition_id,
+ path = %index_path,
+ error = %source,
+ "failed to open sparse index during recovery"
+ );
+ source
+ })?;
+ let first = reader.load_first().await.map_err(|source| {
+ error!(
+ stream_id,
+ topic_id,
+ partition_id,
+ path = %index_path,
+ error = %source,
+ "failed to read first sparse index entry during recovery"
+ );
+ source
+ })?;
+ let last = reader.load_last().await.map_err(|source| {
+ error!(
+ stream_id,
+ topic_id,
+ partition_id,
+ path = %index_path,
+ error = %source,
+ "failed to read last sparse index entry during recovery"
+ );
+ source
+ })?;
+
+ match (first, last) {
+ (Some(first), Some(last)) => {
+ let last_batch_extent = read_batch_extent(messages_path, last.position, messages_size)
+ .ok_or(ServerNgError::RecoveredSegmentSizeDivergence {
+ stream_id,
+ topic_id,
+ partition_id,
+ start_offset,
+ end_offset: last.offset,
+ messages_size_bytes: messages_size,
+ indexed_size_bytes: last.position,
+ })?;
+ if last_batch_extent > messages_size {
+ return Err(ServerNgError::RecoveredSegmentSizeDivergence {
+ stream_id,
+ topic_id,
+ partition_id,
+ start_offset,
+ end_offset: last.offset,
+ messages_size_bytes: messages_size,
+ indexed_size_bytes: last_batch_extent,
+ });
+ }
+ Ok(Some((first.timestamp, last.timestamp, last.offset)))
+ }
+ _ => Ok(None),
+ }
+}
+
+/// End byte of the batch whose command header sits at `position` in the
+/// messages file, or `None` when the header itself does not fit / decode
+/// (`position` past the file, header truncated, or garbage bytes).
+fn read_batch_extent(messages_path: &str, position: u64, messages_size: u64) -> Option<u64> {
+ if position.checked_add(COMMAND_HEADER_SIZE as u64)? > messages_size {
+ return None;
+ }
+ let file = fs::File::open(messages_path).ok()?;
+ let mut header_bytes = [0u8; COMMAND_HEADER_SIZE];
+ file.read_exact_at(&mut header_bytes, position).ok()?;
+ let header = SendMessages2Header::decode(&header_bytes).ok()?;
+ position.checked_add(header.total_size() as u64)
+}
diff --git a/core/server_common/src/indexes_mut.rs b/core/server_common/src/indexes_mut.rs
index 45788796a..b31c59089 100644
--- a/core/server_common/src/indexes_mut.rs
+++ b/core/server_common/src/indexes_mut.rs
@@ -169,9 +169,13 @@ impl IggyIndexesMut {
return None;
}
- Some(IggyIndexView::new(
- &self.buffer[(self.count() - 1) as usize * INDEX_SIZE..],
- ))
+ // Bound to exactly one entry rather than to the buffer end: a file
+ // whose length is not a whole multiple of INDEX_SIZE (e.g. a 24-byte
+ // server-ng sparse index read back through this 16-byte reader on a
+ // mixed-format recovery) would otherwise hand an oversized slice to the
+ // view and trip its length assertion.
+ let start = (self.count() - 1) as usize * INDEX_SIZE;
+ Some(IggyIndexView::new(&self.buffer[start..start + INDEX_SIZE]))
}
/// Finds an index by timestamp using binary search
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 424c71ac3..15629c986 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -1463,9 +1463,13 @@ where
/// Each plane's loopback is dispatched directly to that plane's `on_ack`,
/// avoiding a flat merge that would require re-routing through `on_message`.
///
- /// Invariant: planes do not produce loopback messages for each other.
- /// `on_ack` commits and applies but never calls `push_loopback`, so
- /// draining metadata before partitions is order-independent.
+ /// Invariant: planes do not produce loopback messages FOR EACH OTHER.
+ /// `on_ack` never pushes to another plane's loopback, so draining
+ /// metadata before partitions is order-independent. Within its own
+ /// plane, `on_ack` CAN push loopback entries (a metadata commit promotes
+ /// buffered requests, and each promoted prepare self-acks through
+ /// `send_or_loopback(self)`) -- `repair_primary_self_acks` drains those
+ /// residuals itself; see its interleaved drain.
///
/// # Panics
/// Panics if a loopback message is not a valid `PrepareOk` message.
@@ -1830,6 +1834,11 @@ where
Entry = Message<PrepareHeader>,
Header = PrepareHeader,
>,
+ M: StateMachine<
+ Input = Message<PrepareHeader>,
+ Output = metadata::stm::result::ApplyReply,
+ Error = iggy_common::IggyError,
+ > + StreamsFrontend,
{
let metadata = self.plane.metadata();
let Some(ref consensus) = metadata.consensus else {
@@ -1839,6 +1848,13 @@ where
let actions = consensus.tick(PlaneKind::Metadata);
dispatch_vsr_actions(consensus, metadata.journal.as_ref(), &actions).await;
+
+ // Repair a lost primary self-ack: `RetransmitPrepares` to self is a
+ // no-op, so the timer-driven retransmit above cannot recover the
+ // primary's own missing vote. Without this the commit prefix can pin
+ // forever (commit_min stuck below commit_max). See
+ // `IggyMetadata::repair_primary_self_acks`.
+ metadata.repair_primary_self_acks().await;
}
}