This is an automated email from the ASF dual-hosted git repository.
spetz 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 bb5ed1702 fix(server): preserve write responses and validate CLI input
(#4150)
bb5ed1702 is described below
commit bb5ed1702055f89f784cdf046fe35192fd0d65a4
Author: Hubert Gruszecki <[email protected]>
AuthorDate: Sat Sep 12 22:26:33 2026 +0200
fix(server): preserve write responses and validate CLI input (#4150)
Forwarding can lose durability headers, accepted CLI input can panic
or parse incorrectly, and benchmarks can expose secrets or join the
wrong consumer group.
Preserve response headers, validate CLI input, redact benchmark
metadata and select the matching group. Avoid the telemetry reload
deadlock and align protocol, configuration and command guidance with
the implemented behavior.
---
Cargo.lock | 1 +
README.md | 10 +-
core/bench/Cargo.toml | 1 +
core/bench/src/benchmarks/common.rs | 4 +-
core/bench/src/utils/mod.rs | 14 ++-
core/binary_protocol/src/consensus/header.rs | 43 ++++----
core/binary_protocol/src/consensus/operation.rs | 2 +-
.../src/requests/topics/update_topic.rs | 14 +--
.../src/responses/messages/send_messages.rs | 16 ++-
core/binary_protocol/src/version.rs | 4 +-
core/cli/src/args/consumer_group.rs | 6 +-
core/cli/src/args/consumer_offset.rs | 4 +-
core/cli/src/args/message.rs | 56 +++++++---
core/cli/src/args/mod.rs | 4 +-
core/cli/src/args/personal_access_token.rs | 4 +-
core/cli/src/args/segment.rs | 37 ++++++-
core/cli/src/args/stream.rs | 4 +-
core/cli/src/args/system.rs | 2 +-
core/cli/src/args/topic.rs | 30 +++---
.../src/commands/binary_message/send_messages.rs | 23 ++---
core/common/src/traits/binary_impls/messages.rs | 7 +-
core/common/src/traits/client.rs | 2 +-
core/common/src/types/args/mod.rs | 6 +-
core/common/src/types/message/partitioning.rs | 12 ++-
core/common/src/types/message/partitioning_kind.rs | 4 +-
core/common/src/types/options/mod.rs | 31 ++----
core/consensus/src/impls.rs | 12 +--
core/cpu_allocation/src/lib.rs | 4 +-
core/integration/src/bench_utils.rs | 2 +-
core/integration/tests/bench.rs | 114 +++++++++++++++++++++
.../test_consumer_group_create_command.rs | 10 +-
.../test_consumer_group_help_command.rs | 2 +-
.../test_consumer_offset_get_command.rs | 4 +-
.../test_consumer_offset_set_command.rs | 4 +-
.../tests/cli/general/test_help_command.rs | 8 +-
.../cli/message/test_message_flush_command.rs | 30 +++---
.../tests/cli/message/test_message_help_command.rs | 2 +-
.../tests/cli/message/test_message_send_command.rs | 44 +++++++-
.../test_pat_create_command.rs | 4 +-
.../tests/cli/stream/test_stream_create_command.rs | 6 +-
.../tests/cli/system/test_me_command.rs | 2 +-
.../tests/cli/system/test_snapshot_cmd.rs | 2 +-
.../tests/cli/topic/test_topic_create_command.rs | 16 +--
.../tests/cli/topic/test_topic_update_command.rs | 20 ++--
.../tests/cluster/fast_primary_rejoin.rs | 9 ++
core/integration/tests/mod.rs | 1 +
core/integration/tests/server/mod.rs | 1 +
core/integration/tests/server/telemetry.rs | 63 ++++++++++++
.../tests/server/topic_admission_vsr.rs | 20 ++--
core/metadata/src/stm/snapshot.rs | 30 ++----
core/metadata/src/stm/stream.rs | 99 ++++++++++++++++--
core/sdk/src/vsr.rs | 55 ++++------
core/server/config.toml | 30 +++---
core/server/src/args.rs | 10 +-
core/server/src/dispatch/partition.rs | 2 +-
core/server/src/http/forward.rs | 24 +++--
core/server/src/http/handlers.rs | 2 +-
core/server/src/responses.rs | 22 ++--
core/server/src/shard_allocator.rs | 4 +-
core/server_common/src/consensus_message.rs | 6 +-
core/server_common/src/log/logger.rs | 8 +-
core/server_common/src/memory_pool.rs | 4 +-
core/server_common/src/send_messages.rs | 4 +-
63 files changed, 685 insertions(+), 336 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 4e1263d51..5a191fc13 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -6854,6 +6854,7 @@ dependencies = [
"chrono",
"clap",
"comfy-table",
+ "configs",
"figlet-rs",
"futures-util",
"governor",
diff --git a/README.md b/README.md
index e78873f13..ecd231214 100644
--- a/README.md
+++ b/README.md
@@ -68,14 +68,14 @@ The name is an abbreviation for the Italian Greyhound -
small yet extremely fast
- **Thread per core shared nothing design** together with `io_uring` guarantee
the best possible performance on modern `Linux` systems.
- **Works directly with binary data**, avoiding enforced schema and
serialization/deserialization overhead
- Custom **zero-copy (de)serialization**, which greatly improves the
performance and reduces memory usage.
-- Configurable server features (e.g. caching, segment size, data flush
interval, transport protocols etc.)
+- Configurable server features (e.g. caching and transport protocols), plus
per-topic segment size, durability and flush thresholds
- Server-side storage of **consumer offsets**
- Multiple ways of polling the messages:
- By offset (using the indexes)
- By timestamp (using the time indexes)
- First/Last N messages
- Next N messages for the specific consumer
-- Possibility of **auto committing the offset** (e.g. to achieve
*at-most-once* delivery)
+- Optional **poll auto-commit**; processing guarantees depend on application
processing and offset-commit ordering
- **Consumer groups** providing the message ordering and horizontal scaling
across the connected clients
- **Message expiry** with auto deletion based on the configurable **retention
policy**
- Additional features such as **server side message deduplication**
@@ -89,10 +89,10 @@ The name is an abbreviation for the Italian Greyhound -
small yet extremely fast
are reserved for future disk/network compression support; use message headers
for manual compression today (see
`examples/rust/src/message-headers/message-compression`).
- Optional **data backups and archiving** to disk or **S3** compatible cloud
storage (e.g. AWS S3)
-- Support for **OpenTelemetry** logs & traces + Prometheus metrics
+- Prometheus metrics for the server and connectors runtime, plus
**OpenTelemetry** logs & traces in the connectors runtime. Server OTLP export
is unavailable pending runtime integration.
- Built-in **CLI** to manage the streaming server installable via `cargo
install iggy-cli`
- Built-in **benchmarking app** to test the performance
-- **Single binary deployment** (no external dependencies)
+- **Single binary deployment** without an external broker or database;
dynamically linked builds still require operating-system libraries
- Running as a single node or as a **cluster**, with data replication based on
**[Viewstamped Replication
(VSR)](https://github.com/apache/iggy/blob/master/assets/vsr.pdf)**
@@ -132,7 +132,7 @@ We do also publish edge/dev/nightly releases (e.g.
`0.7.0-edge.1` or `apache/igg
## CLI
-The interactive CLI is implemented under the `cli` project, to provide the
best developer experience. This is a great addition to the Web UI, especially
for all the developers who prefer using the console tools.
+The interactive CLI is implemented under `core/cli`, to provide the best
developer experience. This is a great addition to the Web UI, especially for
all the developers who prefer using the console tools.
Iggy CLI can be installed with `cargo install iggy-cli` and then simply
accessed by typing `iggy` in your terminal.
diff --git a/core/bench/Cargo.toml b/core/bench/Cargo.toml
index 8dc2f65c1..4cbd544ca 100644
--- a/core/bench/Cargo.toml
+++ b/core/bench/Cargo.toml
@@ -39,6 +39,7 @@ charming = { workspace = true }
chrono = { workspace = true }
clap = { workspace = true }
comfy-table = { workspace = true }
+configs = { workspace = true }
figlet-rs = { workspace = true }
futures-util = { workspace = true }
governor = { workspace = true }
diff --git a/core/bench/src/benchmarks/common.rs
b/core/bench/src/benchmarks/common.rs
index ed166bd10..a941e3991 100644
--- a/core/bench/src/benchmarks/common.rs
+++ b/core/bench/src/benchmarks/common.rs
@@ -359,8 +359,8 @@ pub fn build_producing_consumer_groups_futures(
};
let consumer_group_id = if should_consume {
- // Each stream has exactly one CG, server assigns IDs starting
from 0
- Some(start_consumer_group_id)
+ // Match the group name created for this stream.
+ Some(start_consumer_group_id + stream_idx)
} else {
None
};
diff --git a/core/bench/src/utils/mod.rs b/core/bench/src/utils/mod.rs
index 671f6eca8..2b9a1c533 100644
--- a/core/bench/src/utils/mod.rs
+++ b/core/bench/src/utils/mod.rs
@@ -20,6 +20,7 @@ use bench_report::{
numeric_parameter::BenchmarkNumericParameter, params::BenchmarkParams,
transport::BenchmarkTransport,
};
+use configs::{ConfigEnvMappings, server::ServerConfig};
use iggy::prelude::*;
use std::{fs, path::Path};
use tracing::{error, info};
@@ -44,6 +45,13 @@ pub mod cpu_name;
pub mod finish_condition;
pub mod rate_limiter;
+const BENCHMARK_SERVER_ENV_VARS: &[&str] = &[
+ "IGGY_CONFIG_PATH",
+ "IGGY_ENV_PATH",
+ "IGGY_SHARD_RUNTIME_CAPACITY",
+ "IGGY_SHARD_EVENT_INTERVAL",
+];
+
pub fn batch_total_size_bytes(polled_messages: &PolledMessages) -> u64 {
polled_messages
.messages
@@ -187,7 +195,11 @@ fn add_environment_variables(parts: &mut Vec<String>,
server_address: &str) {
if is_localhost {
let iggy_vars: Vec<_> = std::env::vars()
- .filter(|(k, _)| k.starts_with("IGGY_"))
+ .filter(|(name, _)| {
+ BENCHMARK_SERVER_ENV_VARS.contains(&name.as_str())
+ || ServerConfig::find_by_env_name(name)
+ .is_some_and(|mapping| !mapping.is_secret)
+ })
.collect();
if !iggy_vars.is_empty() {
diff --git a/core/binary_protocol/src/consensus/header.rs
b/core/binary_protocol/src/consensus/header.rs
index 233396c18..985c64f8b 100644
--- a/core/binary_protocol/src/consensus/header.rs
+++ b/core/binary_protocol/src/consensus/header.rs
@@ -287,8 +287,8 @@ pub struct RequestHeader {
/// catch a `request` number reused for a different operation: a retry that
/// disagrees with the stamp of the cached reply is refused rather than
/// answered with the wrong reply. Zero means unstamped, which disables the
- /// comparison. The Rust SDK stamps the ops the table dedups; partition and
- /// non-replicated ops, and the other SDKs, leave it zero. The server
+ /// comparison. The Rust SDK stamps metadata/session ops and
`DeleteSegments`;
+ /// partition and non-replicated ops leave it zero. The server
/// verifies any nonzero stamp before routing.
pub request_checksum: u128,
pub timestamp: u64,
@@ -306,14 +306,13 @@ pub struct RequestHeader {
/// cannot restart low after the server drops an entry and the client
/// registers again.
///
- /// Zero on `Register` itself (the client has no epoch to echo yet) and on
- /// sessionless ops; header validation enforces both.
+ /// Header validation requires zero on `Register` itself. `NonReplicated`
+ /// operations also permit zero before a client has registered.
pub session: u64,
- /// Acting user id, stamped by the metadata primary at admission for every
- /// gated client op so the in-apply RBAC gate resolves the same identity on
- /// every replica; on `Register` it carries the freshly authenticated user.
- /// The submitter's wire value is never trusted. Zero for `Logout`,
- /// partition-plane, and server-internal ops.
+ /// Acting user id, stamped by the server for metadata and partition ops
+ /// so every replica uses the authenticated identity for RBAC and dedup.
+ /// On `Register` it carries the freshly authenticated user.
+ /// The submitter's wire value is never trusted.
pub user_id: u32,
pub reserved: [u8; 60],
}
@@ -592,14 +591,12 @@ pub struct ReplyHeader {
/// failure decided before commit (e.g. a dispatch-time authorization
/// denial, or the partition primary rejecting a consumer-offset op).
///
- /// Contract: this is nonzero ONLY on a pre-commit denial, and a deny
- /// reply always carries an EMPTY body. So this header channel and the
- /// committed per-sub-op results in the metadata result section are
mutually
- /// exclusive by construction: a reply either commits (status 0, result
- /// section present) or is denied before commit (status set, no body), and
a
- /// consumer never reconciles the two. Carved from `reserved` exactly like
- /// `user_id` in `RequestHeader` / `PrepareHeader`; no existing field
offset
- /// moves and `validate` does not inspect it.
+ /// Contract: a nonzero status always carries an EMPTY body. With status
+ /// zero, result-framed operations must still decode the result section:
+ /// it can carry a committed result or a pre-commit transient rejection.
+ /// Neither a zero status nor a result section alone proves commitment.
+ /// Carved from `reserved` like `user_id` in `RequestHeader` /
`PrepareHeader`;
+ /// no existing field offset moves and `validate` does not inspect it.
pub status: u32,
pub reserved: [u8; 36],
}
@@ -1041,8 +1038,8 @@ impl ConsensusHeader for PrepareHeader {
/// `checksum` of a prepare no producer sealed.
///
-/// Written by a build predating the identity seal, or by the partition plane.
-/// Verification skips such entries so an older build's WAL still replays.
+/// Written by a build predating the identity seal. Verification skips such
+/// entries so an older build's WAL still replays.
pub const CHECKSUM_UNSEALED: u128 = 0;
/// The frame's body, bounded by `size`. What `checksum_body` covers.
@@ -1385,9 +1382,9 @@ impl ConsensusHeader for StartViewChangeHeader {
}
}
-// DoViewChangeHeader - view change vote (header-only)
+// DoViewChangeHeader - view change vote with log suffix
-/// Replica -> primary candidate: vote for view change. Header-only.
+/// Replica -> replicas: vote for view change, carrying a log-header suffix.
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
#[repr(C)]
pub struct DoViewChangeHeader {
@@ -1559,9 +1556,9 @@ fn suffix_len_of(frame: &str, size: u32) -> Result<usize,
ConsensusError> {
Ok(suffix_len)
}
-// StartViewHeader - new view announcement (header-only)
+// StartViewHeader - new view announcement with log suffix
-/// New primary -> all replicas: start new view. Header-only.
+/// New primary -> replicas: start a new view, with an optional log-header
suffix.
#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
#[repr(C)]
pub struct StartViewHeader {
diff --git a/core/binary_protocol/src/consensus/operation.rs
b/core/binary_protocol/src/consensus/operation.rs
index 8ead254e5..c3e3d2a98 100644
--- a/core/binary_protocol/src/consensus/operation.rs
+++ b/core/binary_protocol/src/consensus/operation.rs
@@ -31,7 +31,7 @@ pub enum Operation {
/// consensus pipeline (prepare/replicate/commit) as normal operations
/// but skips state machine dispatch at commit time, the metadata
/// plane calls `commit_register` directly, which mints the session's
- /// fence epoch (1 at first register, +1 per rebind).
+ /// fence epoch from the committed Register log position.
Register = 1,
/// Non-replicated client request carried in VSR framing. The concrete
diff --git a/core/binary_protocol/src/requests/topics/update_topic.rs
b/core/binary_protocol/src/requests/topics/update_topic.rs
index 902971b37..5f8865a3a 100644
--- a/core/binary_protocol/src/requests/topics/update_topic.rs
+++ b/core/binary_protocol/src/requests/topics/update_topic.rs
@@ -28,16 +28,10 @@ use bytes::BytesMut;
/// `[stream_id:WireIdentifier][topic_id:WireIdentifier][name_len:u8][name:N]
/// [options TLV to end]`
///
-/// Identity and the new name are the only fixed fields; every SETTING rides
the
-/// options block, which mirrors `CreateTopic`'s and carries the same catalog.
-/// A knob added there is updatable here without another layout change, and no
-/// setting has two homes to disagree between.
-///
-/// Keys absent from the block are LEFT ALONE rather than reset to their
-/// defaults. A client built before a key existed cannot send it, so treating
-/// the block as the topic's complete option set would let an old client wipe a
-/// newer knob just by updating the name -- the same forward-compatibility
-/// argument that makes unknown keys survive a round trip.
+/// The options block shares `CreateTopic`'s encoding, but the server validates
+/// which keys can be updated. Supporting a key at creation does not make it
+/// mutable. Absent keys are left unchanged, so updating the name does not
erase
+/// settings introduced after the client was built.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UpdateTopicRequest {
pub stream_id: WireIdentifier,
diff --git a/core/binary_protocol/src/responses/messages/send_messages.rs
b/core/binary_protocol/src/responses/messages/send_messages.rs
index 32785b7be..9ef1d9f47 100644
--- a/core/binary_protocol/src/responses/messages/send_messages.rs
+++ b/core/binary_protocol/src/responses/messages/send_messages.rs
@@ -38,13 +38,9 @@ const CONFIRMATION_SIZE: usize = 20;
/// ```
///
/// `base_offset` is the offset assigned to the first message of the batch in
-/// that partition, bounded by three properties of the send path:
-/// - Delivery is at-least-once. An earlier retry of the same batch may already
-/// have committed at a lower offset, so the value never implies uniqueness.
-/// - A batch is confirmed once it is committed in memory, not once it is
-/// fsynced. A crash-restart can stamp a later batch with an offset a client
-/// has already recorded.
-/// - The legacy server confirms nothing, so its confirmation list is empty.
+/// that partition. It does not imply uniqueness: retries outside the server's
+/// deduplication coverage may append the batch again. Crash durability depends
+/// on the topic's durability policy, not on the presence of offset
information.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SendMessagesConfirmationResponse {
pub stream_id: u32,
@@ -91,9 +87,9 @@ impl WireDecode for SendMessagesConfirmationResponse {
/// [confirmations_count:4][SendMessagesConfirmationResponse]*
/// ```
///
-/// `confirmations_count == 0` means the batch committed with no offsets to
-/// report. The legacy server goes further and answers a successful send with
no
-/// body at all, so a caller sees an empty list either way and must handle it.
+/// `confirmations_count == 0` supplies no offset information. Callers must
also
+/// handle an empty successful reply body, which the server uses for a request
+/// classified as a duplicate. Neither response identifies an append position.
///
/// The server currently reports a single partition per request; the list
/// decodes any count, so a later multi-partition send needs no wire change.
diff --git a/core/binary_protocol/src/version.rs
b/core/binary_protocol/src/version.rs
index 6d26df2fd..2b158c249 100644
--- a/core/binary_protocol/src/version.rs
+++ b/core/binary_protocol/src/version.rs
@@ -47,8 +47,8 @@
//!
//! Integer order equals semver order. The value tracks the
//! `iggy_binary_protocol` crate release; under 0.x a minor bump may break
-//! the wire, so the gate is minor-scoped. Past 1.0.0 the gate follows
-//! strict semver: major bump = incompatible, minor/patch = compatible.
+//! the wire, so the gate is minor-scoped. Compatibility across minor versions
+//! after 1.0.0 requires changing the minimum-version calculation below.
//!
//! ## `ClientVersionInfo` body prefix
//!
diff --git a/core/cli/src/args/consumer_group.rs
b/core/cli/src/args/consumer_group.rs
index c37548348..488511f62 100644
--- a/core/cli/src/args/consumer_group.rs
+++ b/core/cli/src/args/consumer_group.rs
@@ -21,11 +21,11 @@ use iggy::prelude::Identifier;
#[derive(Debug, Clone, Subcommand)]
pub(crate) enum ConsumerGroupAction {
- /// Create consumer group with given ID and name for given stream ID and
topic ID.
+ /// Create consumer group with given name for given stream ID and topic ID.
///
/// Stream ID can be specified as a stream name or ID
/// Topic ID can be specified as a topic name or ID
- /// If group ID is not provided then the server will automatically assign
it
+ /// The server assigns the group ID. The legacy --group-id flag is ignored.
///
/// Examples:
/// iggy consumer-group create 1 1 prod
@@ -94,7 +94,7 @@ pub(crate) struct ConsumerGroupCreateArgs {
/// Topic ID can be specified as a topic name or ID
#[arg(value_parser = clap::value_parser!(Identifier))]
pub(crate) topic_id: Identifier,
- /// Consumer group ID to create
+ /// Legacy consumer group ID flag (ignored)
#[clap(short, long)]
pub(crate) group_id: Option<u32>,
/// Consumer group name to create
diff --git a/core/cli/src/args/consumer_offset.rs
b/core/cli/src/args/consumer_offset.rs
index 1e31b411a..40241c560 100644
--- a/core/cli/src/args/consumer_offset.rs
+++ b/core/cli/src/args/consumer_offset.rs
@@ -78,7 +78,7 @@ pub(crate) struct ConsumerOffsetGetArgs {
/// Partitions ID for which consumer offset is retrieved
#[arg(value_parser = clap::value_parser!(u32).range(0..))]
pub(crate) partition_id: u32,
- /// Consumer kind: "consumer" for regular consumer, "consumer_group" for
consumer group
+ /// Consumer kind: "consumer" for regular consumer, "consumer-group" for
consumer group
#[arg(short = 'k', long = "kind", default_value = "consumer", value_enum)]
pub(crate) kind: ConsumerKind,
}
@@ -106,7 +106,7 @@ pub(crate) struct ConsumerOffsetSetArgs {
pub(crate) partition_id: u32,
/// Offset to set
pub(crate) offset: u64,
- /// Consumer kind: "consumer" for regular consumer, "consumer_group" for
consumer group
+ /// Consumer kind: "consumer" for regular consumer, "consumer-group" for
consumer group
#[arg(short = 'k', long = "kind", default_value = "consumer", value_enum)]
pub(crate) kind: ConsumerKind,
}
diff --git a/core/cli/src/args/message.rs b/core/cli/src/args/message.rs
index 74c12b4b5..f435fd610 100644
--- a/core/cli/src/args/message.rs
+++ b/core/cli/src/args/message.rs
@@ -46,12 +46,11 @@ pub(crate) enum MessageAction {
/// iggy message poll --offset 0 stream topic 1
#[clap(verbatim_doc_comment, visible_alias = "p")]
Poll(PollMessagesArgs),
- /// Flush messages from given topic ID and given stream ID
+ /// Legacy message flush command (unsupported by the server)
///
- /// Command is used to force a flush of unsaved_buffer to disk
- /// for specific stream, topic and partition. If fsync is enabled
- /// then the data is flushed to disk and fsynced, otherwise the
- /// data is only flushed to disk.
+ /// Binary transports return FeatureUnavailable. HTTP has no flush route.
+ /// For acknowledgements backed by stable storage, create the topic with
+ /// --durability persisted.
///
/// Stream ID can be specified as a stream name or ID
/// Topic ID can be specified as a topic name or ID
@@ -82,9 +81,9 @@ pub(crate) struct SendMessagesArgs {
pub(crate) partition_id: Option<u32>,
/// Messages key which will be used to partition the messages
///
- /// Value of the key will be used by the server to calculate the partition
ID
+ /// The key must contain 1 to 255 bytes. Binary clients resolve the
partition ID; HTTP resolves it on the server.
#[clap(verbatim_doc_comment)]
- #[clap(short, long, group = "partitioning")]
+ #[clap(short, long, value_parser = parse_message_key, group =
"partitioning")]
pub(crate) message_key: Option<String>,
/// Messages to be sent
///
@@ -117,6 +116,10 @@ pub(crate) struct SendMessagesArgs {
pub(crate) input_file: Option<String>,
}
+fn parse_message_key(value: &str) -> Result<String, IggyError> {
+ Partitioning::messages_key_str(value).map(|_| value.to_owned())
+}
+
/// Parse Header Key, Kind and Value from the string separated by a ':'
fn parse_key_val(s: &str) -> Result<(HeaderKey, HeaderValue), IggyError> {
let parts = s.splitn(3, ':').collect::<Vec<_>>();
@@ -265,23 +268,22 @@ pub(crate) struct PollMessagesArgs {
#[derive(Debug, Clone, Args)]
pub(crate) struct FlushMessagesArgs {
- /// ID of the stream for which messages will be flushed
+ /// Stream ID for the flush request
///
/// Stream ID can be specified as a stream name or ID
#[arg(value_parser = clap::value_parser!(Identifier))]
pub(crate) stream_id: Identifier,
- /// ID of the topic for which messages will be flushed
+ /// Topic ID for the flush request
///
/// Topic ID can be specified as a topic name or ID
#[arg(value_parser = clap::value_parser!(Identifier))]
pub(crate) topic_id: Identifier,
- /// Partition ID for which messages will be flushed
+ /// Partition ID for the flush request
#[arg(value_parser = clap::value_parser!(u32).range(0..))]
pub(crate) partition_id: u32,
- /// fsync flushed data to disk
+ /// Request fsync (unsupported by the server)
///
- /// If option is enabled then the data is flushed to disk and fsynced,
- /// otherwise the data is only flushed to disk. Default is false.
+ /// The server rejects flush requests regardless of this flag.
#[clap(verbatim_doc_comment)]
#[clap(short, long, default_value_t = false)]
pub(crate) fsync: bool,
@@ -290,8 +292,36 @@ pub(crate) struct FlushMessagesArgs {
#[cfg(test)]
mod tests {
use super::*;
+ use crate::args::{Command, IggyConsoleArgs};
+ use clap::Parser;
use std::str::FromStr;
+ #[test]
+ fn given_valid_message_key_when_sending_should_preserve_key_bytes() {
+ let max_key_bytes = usize::from(u8::MAX);
+ for key in [
+ "x".to_owned(),
+ "x".repeat(max_key_bytes),
+ format!("{}x", "é".repeat(max_key_bytes / "é".len())),
+ ] {
+ let parsed = IggyConsoleArgs::try_parse_from([
+ "iggy",
+ "message",
+ "send",
+ "--message-key",
+ &key,
+ "stream",
+ "topic",
+ "payload",
+ ])
+ .unwrap();
+ let Some(Command::Message(MessageAction::Send(args))) =
parsed.command else {
+ panic!("Expected the message send command");
+ };
+ assert_eq!(args.message_key.as_deref(), Some(key.as_str()));
+ }
+ }
+
#[test]
fn parse_key_val_should_parse_string() {
let expected_value: &str = "value";
diff --git a/core/cli/src/args/mod.rs b/core/cli/src/args/mod.rs
index 42d517810..589e45d53 100644
--- a/core/cli/src/args/mod.rs
+++ b/core/cli/src/args/mod.rs
@@ -87,7 +87,7 @@ pub(crate) struct IggyConsoleArgs {
#[derive(Debug, Parser)]
#[command(author, version, about, long_about = None)]
pub(crate) struct CliOptions {
- /// Quiet mode (disabled stdout printing)
+ /// Suppress logging output (command-specific output may remain)
#[clap(short, long, default_value_t = false)]
pub(crate) quiet: bool,
@@ -159,7 +159,7 @@ pub(crate) enum Command {
/// get current client info
///
/// Command connects to Iggy server and collects client info like client
ID, user ID
- /// server address and protocol type.
+ /// client address as seen by the server and protocol type.
#[clap(verbatim_doc_comment)]
Me,
/// list the options a resource's create command accepts
diff --git a/core/cli/src/args/personal_access_token.rs
b/core/cli/src/args/personal_access_token.rs
index cd56425f0..30944e9e8 100644
--- a/core/cli/src/args/personal_access_token.rs
+++ b/core/cli/src/args/personal_access_token.rs
@@ -25,7 +25,7 @@ pub(crate) enum PersonalAccessTokenAction {
///
/// Create personal access token which allow authenticating the clients
using
/// a token, instead of the regular credentials (username and password)
- /// In quiet mode only the personal access token name is printed
+ /// In quiet mode the raw token is printed unless --store-token is set
///
/// Examples
/// iggy pat create name
@@ -63,7 +63,7 @@ pub(crate) struct PersonalAccessTokenCreateArgs {
/// Generated token is stored in a platform-specific secure storage
without revealing
/// its content to the user. It can be used to authenticate on iggy server
using
/// associated name and -n/--token-name command line option instead of
-u/--username
- /// and -p/--password or -t/--token. In quiet mode only the token name is
printed.
+ /// and -p/--password or -t/--token. Quiet mode prints no token or
confirmation.
/// This option can only be used for creating tokens which does not have
expiry time set.
#[clap(short, long, default_value_t = false, group = "store")]
pub(crate) store_token: bool,
diff --git a/core/cli/src/args/segment.rs b/core/cli/src/args/segment.rs
index 948e9b85c..68cce4d23 100644
--- a/core/cli/src/args/segment.rs
+++ b/core/cli/src/args/segment.rs
@@ -25,7 +25,7 @@ pub(crate) enum SegmentAction {
///
/// Stream ID can be specified as a stream name or ID
/// Topic ID can be specified as a topic name or ID
- /// Partition ID can be specified as a name or ID
+ /// Partition ID must be numeric
///
/// Examples
/// iggy segment delete 1 1 1 10
@@ -49,9 +49,42 @@ pub(crate) struct SegmentDeleteArgs {
#[arg(value_parser = clap::value_parser!(Identifier))]
pub(crate) topic_id: Identifier,
/// Partition ID to delete segments
- #[arg(value_parser = clap::value_parser!(Identifier))]
pub(crate) partition_id: u32,
/// Segments count to be deleted
#[arg(value_parser = clap::value_parser!(u32).range(1..100_001))]
pub(crate) segments_count: u32,
}
+
+#[cfg(test)]
+mod tests {
+ use super::SegmentAction;
+ use crate::args::{Command, IggyConsoleArgs};
+ use clap::Parser;
+
+ #[test]
+ fn given_numeric_partition_when_deleting_segments_should_parse_command() {
+ for partition in ["0", "1"] {
+ let parsed = IggyConsoleArgs::try_parse_from([
+ "iggy", "segment", "delete", "dev", "events", partition, "3",
+ ])
+ .unwrap();
+ let Some(Command::Segment(SegmentAction::Delete(args))) =
parsed.command else {
+ panic!("Expected the segment delete command");
+ };
+ assert_eq!(args.partition_id.to_string(), partition);
+ assert_eq!(args.segments_count, 3);
+ }
+ assert!(
+ IggyConsoleArgs::try_parse_from([
+ "iggy",
+ "segment",
+ "delete",
+ "dev",
+ "events",
+ "named-partition",
+ "3",
+ ])
+ .is_err()
+ );
+ }
+}
diff --git a/core/cli/src/args/stream.rs b/core/cli/src/args/stream.rs
index 56eff8b30..5e8f6a00a 100644
--- a/core/cli/src/args/stream.rs
+++ b/core/cli/src/args/stream.rs
@@ -23,7 +23,7 @@ use iggy::prelude::Identifier;
pub(crate) enum StreamAction {
/// Create stream with given name
///
- /// If stream ID is not provided then the server will automatically assign
it
+ /// The server assigns the stream ID. The legacy --stream-id flag is
ignored.
///
/// Examples:
/// iggy stream create prod
@@ -79,7 +79,7 @@ pub(crate) enum StreamAction {
#[derive(Debug, Clone, Args)]
pub(crate) struct StreamCreateArgs {
- /// Stream ID to create
+ /// Legacy stream ID flag (ignored)
#[clap(short, long)]
pub(crate) stream_id: Option<u32>,
/// Name of the stream
diff --git a/core/cli/src/args/system.rs b/core/cli/src/args/system.rs
index a7195a2f1..a0c23dad1 100644
--- a/core/cli/src/args/system.rs
+++ b/core/cli/src/args/system.rs
@@ -67,7 +67,7 @@ pub(crate) struct SnapshotArgs {
///
/// Examples:
/// - `--compression bzip2` for higher compression.
- /// - `--compression none` to store without compression.
+ /// - `--compression stored` to store without compression.
#[arg(verbatim_doc_comment, short, long, value_parser =
clap::value_parser!(SnapshotCompression))]
pub(crate) compression: Option<SnapshotCompression>,
diff --git a/core/cli/src/args/topic.rs b/core/cli/src/args/topic.rs
index b5447ef66..6d3d83838 100644
--- a/core/cli/src/args/topic.rs
+++ b/core/cli/src/args/topic.rs
@@ -24,7 +24,7 @@ pub(crate) enum TopicAction {
/// Create topic with given name, number of partitions, compression
algorithm and expiry time for given stream ID
///
/// Stream ID can be specified as a stream name or ID
- /// If topic ID is not provided then the server will automatically assign
it
+ /// The server assigns the topic ID. The legacy --topic-id flag is ignored.
///
/// Examples
/// iggy topic create 1 sensor1 2 gzip 15days
@@ -51,11 +51,11 @@ pub(crate) enum TopicAction {
/// Topic ID can be specified as a topic name or ID
///
/// Examples
- /// iggy update 1 1 sensor3 none
- /// iggy update prod sensor3 old-sensor none
- /// iggy update test debugs ready gzip 15days
- /// iggy update 1 1 new-name gzip
- /// iggy update 1 2 new-name none 1day 1hour 1min 1sec
+ /// iggy topic update 1 1 sensor3 none
+ /// iggy topic update prod sensor3 old-sensor none
+ /// iggy topic update test debugs ready gzip 15days
+ /// iggy topic update 1 1 new-name gzip
+ /// iggy topic update 1 2 new-name none 1day 1hour 1min 1sec
#[clap(verbatim_doc_comment, visible_alias = "u")]
Update(TopicUpdateArgs),
/// Get topic detail for given topic ID and stream ID
@@ -103,23 +103,23 @@ pub(crate) struct TopicCreateArgs {
pub(crate) stream_id: Identifier,
/// Name of the topic
pub(crate) name: String,
- /// Topic ID to create
+ /// Legacy topic ID flag (ignored)
#[clap(short, long)]
pub(crate) topic_id: Option<u32>,
/// Number of partitions inside the topic
pub(crate) partitions_count: u32,
- /// Compression algorithm for the topic, set to "none" for no compression
+ /// Compression metadata (none or gzip). Payload compression is not
implemented
#[arg(value_parser = clap::value_parser!(CompressionAlgorithm),
verbatim_doc_comment)]
pub(crate) compression_algorithm: CompressionAlgorithm,
/// Max topic size in human-readable format like "unlimited" or "15GB"
///
- /// "server_default" or skipping parameter makes CLI to use server default
(from current server config) max topic size
- /// Can't be lower than segment size in the config.
+ /// Skipping this parameter or using "server_default" creates a topic with
unlimited size.
+ /// A finite size cannot be lower than the topic segment size.
#[arg(short, long, default_value = "server_default", verbatim_doc_comment)]
pub(crate) max_topic_size: MaxTopicSize,
/// Message expiry time in human-readable format like "unlimited" or
"15days 2min 2s"
///
- /// "server_default" or skipping parameter makes CLI to use server default
(from current server config) expiry time
+ /// Skipping this parameter or using "server_default" creates a topic with
no message expiry.
#[arg(default_value = "server_default", value_parser =
clap::value_parser!(IggyExpiry), verbatim_doc_comment)]
pub(crate) message_expiry: Vec<IggyExpiry>,
/// Message completion policy: replicated or persisted. Both policies
store messages on disk.
@@ -166,18 +166,18 @@ pub(crate) struct TopicUpdateArgs {
pub(crate) topic_id: Identifier,
/// New name for the topic
pub(crate) name: String,
- /// Compression algorithm for the topic, set to "none" for no compression
+ /// Compression metadata (none or gzip). Payload compression is not
implemented
#[arg(value_parser = clap::value_parser!(CompressionAlgorithm),
verbatim_doc_comment)]
pub(crate) compression_algorithm: CompressionAlgorithm,
/// New max topic size in human-readable format like "unlimited" or "15GB"
///
- /// "server_default" or skipping parameter makes CLI to use server default
(from current server config) max topic size
- /// Can't be lower than segment size in the config.
+ /// Skipping this parameter or using "server_default" preserves the
current max topic size.
+ /// A finite size cannot be lower than the topic segment size.
#[arg(short, long, default_value = "server_default", verbatim_doc_comment)]
pub(crate) max_topic_size: MaxTopicSize,
/// New message expiry time in human-readable format like "unlimited" or
"15days 2min 2s"
///
- /// "server_default" or skipping parameter makes CLI to use server default
(from current server config) expiry time
+ /// Skipping this parameter or using "server_default" preserves the
current message expiry.
#[arg(default_value = "server_default", value_parser =
clap::value_parser!(IggyExpiry), verbatim_doc_comment)]
pub(crate) message_expiry: Vec<IggyExpiry>,
}
diff --git a/core/cli/src/commands/binary_message/send_messages.rs
b/core/cli/src/commands/binary_message/send_messages.rs
index 01da3474e..ddb61b831 100644
--- a/core/cli/src/commands/binary_message/send_messages.rs
+++ b/core/cli/src/commands/binary_message/send_messages.rs
@@ -29,7 +29,8 @@ use tracing::{Level, event};
pub struct SendMessagesCmd {
stream_id: Identifier,
topic_id: Identifier,
- partitioning: Partitioning,
+ partition_id: Option<u32>,
+ message_key: Option<String>,
messages: Option<Vec<String>>,
headers: Vec<(HeaderKey, HeaderValue)>,
input_file: Option<String>,
@@ -45,19 +46,11 @@ impl SendMessagesCmd {
headers: Vec<(HeaderKey, HeaderValue)>,
input_file: Option<String>,
) -> Self {
- let partitioning = match (partition_id, message_key) {
- (Some(_), Some(_)) => unreachable!(),
- (Some(partition_id), None) =>
Partitioning::partition_id(partition_id),
- (None, Some(message_key)) =>
Partitioning::messages_key_str(message_key.as_str())
- .unwrap_or_else(|_| {
- panic!("Failed to create Partitioning with {message_key}
string message key")
- }),
- (None, None) => Partitioning::default(),
- };
Self {
stream_id,
topic_id,
- partitioning,
+ partition_id,
+ message_key,
messages,
headers,
input_file,
@@ -90,6 +83,12 @@ impl CliCommand for SendMessagesCmd {
}
async fn execute_cmd(&mut self, client: &dyn Client) -> anyhow::Result<(),
anyhow::Error> {
+ let partitioning = match (self.partition_id,
self.message_key.as_deref()) {
+ (Some(_), Some(_)) => unreachable!(),
+ (Some(partition_id), None) =>
Partitioning::partition_id(partition_id),
+ (None, Some(message_key)) =>
Partitioning::messages_key_str(message_key)?,
+ (None, None) => Partitioning::default(),
+ };
let mut messages = if let Some(input_file) = &self.input_file {
let mut file = tokio::fs::OpenOptions::new()
.read(true)
@@ -165,7 +164,7 @@ impl CliCommand for SendMessagesCmd {
.send_messages(
&self.stream_id,
&self.topic_id,
- &self.partitioning,
+ &partitioning,
&mut messages,
)
.await
diff --git a/core/common/src/traits/binary_impls/messages.rs
b/core/common/src/traits/binary_impls/messages.rs
index d8b12410d..f087bc7b1 100644
--- a/core/common/src/traits/binary_impls/messages.rs
+++ b/core/common/src/traits/binary_impls/messages.rs
@@ -268,10 +268,9 @@ pub fn decode_send_confirmations(response: &[u8]) ->
Result<SendMessagesResponse
///
/// An unreadable body degrades to no confirmations instead of an error. The
/// producer retry loop filters nothing and resends on any `Err`, so failing
-/// here would turn one committed write into as many copies as the retry budget
-/// allows, on a plane that keeps no reply cache to deduplicate them. Reporting
-/// a zeroed entry instead would be no better: the caller cannot tell it from a
-/// genuine commit at offset 0 and would checkpoint the shape mismatch.
+/// here would resend a committed write under a new request id, outside the
+/// partition's retry deduplication. A zeroed entry would be indistinguishable
+/// from a genuine commit at offset 0 and would checkpoint the shape mismatch.
fn committed_send_confirmations(response: &[u8]) -> SendMessagesResponse {
decode_send_confirmations(response).unwrap_or_else(|_|
SendMessagesResponse {
confirmations: Vec::new(),
diff --git a/core/common/src/traits/client.rs b/core/common/src/traits/client.rs
index 149c3cb4c..9c6f4fa17 100644
--- a/core/common/src/traits/client.rs
+++ b/core/common/src/traits/client.rs
@@ -26,7 +26,7 @@ use std::fmt::Debug;
/// The client trait which is the main interface to the Iggy server.
/// It consists of multiple modules, each of which is responsible for a
specific set of commands.
-/// Except the ping, login and get me, all the other methods require
authentication.
+/// Server operations require authentication, except ping and the login flows.
#[async_trait]
pub trait Client:
ClusterClient
diff --git a/core/common/src/types/args/mod.rs
b/core/common/src/types/args/mod.rs
index d44b223ce..4dc5ff389 100644
--- a/core/common/src/types/args/mod.rs
+++ b/core/common/src/types/args/mod.rs
@@ -160,14 +160,14 @@ pub struct ArgsOptional {
/// The optional send window for QUIC
///
- /// [default: 100000]
+ /// [default: 1000000]
#[arg(long)]
#[serde(skip_serializing_if = "Option::is_none")]
pub quic_send_window: Option<u64>,
/// The optional receive window for QUIC
///
- /// [default: 100000]
+ /// [default: 1000000]
#[arg(long)]
#[serde(skip_serializing_if = "Option::is_none")]
pub quic_receive_window: Option<u64>,
@@ -188,7 +188,7 @@ pub struct ArgsOptional {
/// The optional maximum idle timeout for QUIC
///
- /// [default: 10000]
+ /// [default: 100000]
#[arg(long)]
#[serde(skip_serializing_if = "Option::is_none")]
pub quic_max_idle_timeout: Option<u64>,
diff --git a/core/common/src/types/message/partitioning.rs
b/core/common/src/types/message/partitioning.rs
index 615aa8ffb..38168f573 100644
--- a/core/common/src/types/message/partitioning.rs
+++ b/core/common/src/types/message/partitioning.rs
@@ -25,14 +25,18 @@ use std::{
hash::{Hash, Hasher},
};
-/// A type that defines a what strategy the server should choose to partition
the messages.
+/// Selects the partitioning strategy for a batch of messages.
///
/// Iggy uses a hierarchical model for append-only logs. A stream contains
topics which hold partitions. Each partition is an append-only log.[^note]
/// A producer of messages such as an `IggyProducer`, that appends messages to
the log can choose between three partitioning strategies.
-/// - `Balanced` - the partition ID is calculated by the server using the
round-robin algorithm.
-/// - `MessagesKey` - the partition ID is calculated by the server using the
hash of the provided messages key.
+/// - `Balanced` - selects a partition using round-robin.
+/// - `MessagesKey` - hashes the key modulo the topic partition count.
/// - `PartitionId` - the partition ID is provided by the client.
///
+/// Binary clients resolve `Balanced` and `MessagesKey` before sending. The
HTTP
+/// client sends the strategy to the server, which resolves it at admission.
+/// Changing the partition count can change a key's destination.
+///
/// Note, that using a [`Partitioner`] on top of [`Partitioning`] sets the
strategy to [`PartitioningKind::PartitionId`]. The value is then computed
/// based on your concrete implementation of
[`Partitioner::calculate_partition_id()`].
///
@@ -72,7 +76,7 @@ impl Display for Partitioning {
}
impl Partitioning {
- /// Partition the messages using the balanced round-robin algorithm on the
server.
+ /// Partition the messages using round-robin.
pub fn balanced() -> Self {
Partitioning {
kind: PartitioningKind::Balanced,
diff --git a/core/common/src/types/message/partitioning_kind.rs
b/core/common/src/types/message/partitioning_kind.rs
index 205d2b882..b2c538f8d 100644
--- a/core/common/src/types/message/partitioning_kind.rs
+++ b/core/common/src/types/message/partitioning_kind.rs
@@ -26,12 +26,12 @@ use std::{
#[derive(Debug, Serialize, Deserialize, Eq, PartialEq, Default, Copy, Clone)]
#[serde(rename_all = "snake_case")]
pub enum PartitioningKind {
- /// The partition ID is calculated by the server using the round-robin
algorithm.
+ /// Select a partition using round-robin.
#[default]
Balanced,
/// The partition ID is provided by the client.
PartitionId,
- /// The partition ID is calculated by the server using the hash of the
provided messages key.
+ /// Hash the messages key modulo the topic partition count.
MessagesKey,
}
diff --git a/core/common/src/types/options/mod.rs
b/core/common/src/types/options/mod.rs
index 38e80366e..717f6b0c4 100644
--- a/core/common/src/types/options/mod.rs
+++ b/core/common/src/types/options/mod.rs
@@ -229,15 +229,9 @@ pub mod topic_option_keys {
/// Values an absent topic option resolves to at admission.
///
-/// These are the knobs' single source of truth: they used to live in
-/// topic creation options,
-/// which meant every one of them had two homes and an operator could not tell
-/// which won. A topic carries whatever it was created with; anything the
-/// client did not send resolves to the constant here and is persisted as a
-/// derived entry, so the effective value is always visible on `GetTopic`.
-///
-/// Each value matches what the shipped `config.toml` carried, so removing the
-/// keys changed no behavior for a topic created without options.
+/// These defaults are fixed by the option catalog, not server configuration.
+/// Values the client omits are resolved at admission and persisted as derived
+/// entries, so `GetTopic` reports the effective values and their provenance.
pub const DEFAULT_PARTITIONS_COUNT: u32 = 1;
/// `MaxTopicSize::Unlimited` (was `[topic] max_size = "unlimited"`).
pub const DEFAULT_MAX_TOPIC_SIZE: u64 = u64::MAX;
@@ -249,14 +243,10 @@ pub const DEFAULT_SEGMENT_SIZE: u64 = 1024 * 1024 * 1024;
pub const DEFAULT_MESSAGES_REQUIRED_TO_SAVE: u32 = 1024;
/// Preallocation is opt-in.
///
-/// That default was never actually in force: the reservation ran through
-/// `compio::spawn_blocking`, which panics the shard because shard executors
-/// disable the blocking pool, so any deployment that worked at all had
-/// preallocation off. With the call fixed to run inline it reserves real
-/// extents, and `FALLOC_FL_KEEP_SIZE` against the 1 GiB default segment size
-/// means 1 GiB of disk per partition the moment it is created -- a full test
-/// sweep reserved 393 GB before this was flipped. A topic that wants the
-/// latency benefit asks for it with `preallocate_segments`.
+/// Each owned partition requests `segment_size` bytes when opening a segment.
+/// Linux uses `FALLOC_FL_KEEP_SIZE` to reserve extents without extending the
+/// logical file. Unsupported or failed reservations fall back to ordinary
+/// allocation with a warning.
pub const DEFAULT_PREALLOCATE_SEGMENTS: bool = false;
/// 1 MiB (was `[partition] size_of_messages_required_to_save`).
pub const DEFAULT_SIZE_OF_MESSAGES_REQUIRED_TO_SAVE: u64 = 1024 * 1024;
@@ -347,10 +337,9 @@ pub fn validate_preallocated_topic_bytes(
/// Validate an explicit per-topic `segment_size` against its bounds.
///
-/// `ceiling` is node-derived: the smaller of the global segment maximum and
-/// the state-transfer artifact budget minus one bus frame (a segment may
-/// close one whole batch past its cap; an artifact ceiling below that
-/// refuses a legal segment and livelocks the partition's rejoin).
+/// Admission uses [`MAX_TOPIC_SEGMENT_SIZE`] as `ceiling`. Server startup
+/// separately validates that state-transfer budgets can hold a segment at
+/// that ceiling plus one whole batch of overshoot.
///
/// # Errors
///
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index 649843209..f445d25dd 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -1657,8 +1657,8 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
self.commit_min.set(commit_min);
}
- /// Maximum number of faulty replicas that can be tolerated.
- /// For a cluster of 2f+1 replicas, this returns f.
+ /// `StartViewChange` votes needed from other replicas before sending
`DoViewChange`.
+ /// This is f for 2f+1 replicas, not the failure bound for every quorum
configuration.
#[must_use]
pub const fn max_faulty(&self) -> usize {
(self.replica_count as usize - 1) / 2
@@ -4042,8 +4042,8 @@ where
// cost on the produce path, and it would describe the WRONG bytes:
// `stamp_prepare_for_persistence` rewrites the command header INSIDE
this
// sealed region before the entry is journaled. Leaving those prepares
at `0`
- // is the designed "nothing to verify" sentinel, so a future durable
partition
- // journal skips verification instead of failing every entry as
corrupt.
+ // is the designed "nothing to verify" sentinel: partition prepares
+ // leave message-body integrity to the batch checksums.
//
// TODO(consensus): a partition prepare's `checksum` covers its header
alone,
// so two at one op with matching header fields are indistinguishable
however
@@ -4053,8 +4053,8 @@ where
// the zero. Two closures, both larger than they look:
//
// 1. The batch checksum, recomputed after
`stamp_prepare_for_persistence`.
- // But stamping runs per replica after replication and folds
`base_offset`
- // in, so identity would change at stamp time and the journaled
entry would
+ // But stamping runs after this projection and folds `base_offset`
in,
+ // so identity would change at stamp time and the journaled entry
would
// no longer match the pipeline entry `handle_prepare_ok` compares.
// 2. The stamp-invariant cover: everything past the 256-byte command
header,
// which stamping never touches. Identical on every replica, safe
to seal
diff --git a/core/cpu_allocation/src/lib.rs b/core/cpu_allocation/src/lib.rs
index 48719f889..a8776bd33 100644
--- a/core/cpu_allocation/src/lib.rs
+++ b/core/cpu_allocation/src/lib.rs
@@ -36,8 +36,8 @@ pub use allowed_cpus::allowed_cpus;
/// Tell server how many CPU cores to grab for shards, and how.
///
/// Server make one shard per core. This say which cores. Pick one:
-/// - `All`: take every core machine have.
-/// - `Count(n)`: take first `n` cores.
+/// - `All`: use the parallelism available to the process.
+/// - `Count(n)`: use `n` shards, pinned within the allowed CPU set when
enabled.
/// - `Range(a, b)`: take cores `a` up to (not including) `b`.
/// - `NumaAware(..)`: smart pick by NUMA node, keep memory close to core.
///
diff --git a/core/integration/src/bench_utils.rs
b/core/integration/src/bench_utils.rs
index dcfbde1c8..561e6e7a3 100644
--- a/core/integration/src/bench_utils.rs
+++ b/core/integration/src/bench_utils.rs
@@ -37,7 +37,7 @@ const DEFAULT_NUMBER_OF_STREAMS: u64 = 8;
// the stale-binary hint below with it. Exists because a stale prebuilt
// iggy-bench speaking an outdated protocol hangs both sides silently
// instead of erroring.
-const BENCH_WAIT_TIMEOUT: Duration = Duration::from_secs(240);
+pub const BENCH_WAIT_TIMEOUT: Duration = Duration::from_secs(240);
pub fn run_bench_and_wait_for_finish(
server_addr: &str,
diff --git a/core/integration/tests/bench.rs b/core/integration/tests/bench.rs
new file mode 100644
index 000000000..f9a9bc19d
--- /dev/null
+++ b/core/integration/tests/bench.rs
@@ -0,0 +1,114 @@
+// 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 std::{fs, process::Command};
+
+use assert_cmd::prelude::CommandCargoExt;
+use iggy::prelude::*;
+use iggy_common::TransportProtocol;
+use integration::bench_utils::{BENCH_WAIT_TIMEOUT,
run_bench_and_wait_for_finish};
+use integration::harness::{TestHarness, TestServerConfig};
+use serial_test::parallel;
+
+const BENCHMARK_DATA_BYTES: u64 = 5_000_000;
+const REPORT_TEST_DATA: &str = "1MiB";
+const TEST_SECRET: &str = "sensitive-benchmark-test-value";
+
+#[tokio::test]
+#[parallel]
+async fn
given_fresh_server_when_running_end_to_end_group_benchmark_should_finish() {
+ let mut harness = TestHarness::builder()
+ .cluster_nodes(1)
+ .server(TestServerConfig::default())
+ .build()
+ .unwrap();
+ harness.start().await.unwrap();
+
+ run_bench_and_wait_for_finish(
+ &harness.server().raw_tcp_addr().unwrap(),
+ &TransportProtocol::Tcp,
+ "end-to-end-producing-consumer-group",
+ IggyByteSize::new(BENCHMARK_DATA_BYTES),
+ );
+}
+
+#[tokio::test]
+#[parallel]
+async fn
given_secret_environment_when_saving_benchmark_should_omit_credentials() {
+ let mut harness = TestHarness::builder()
+ .cluster_nodes(1)
+ .server(TestServerConfig::default())
+ .build()
+ .unwrap();
+ harness.start().await.unwrap();
+ let output_dir = tempfile::tempdir().unwrap();
+ let server_address = harness.server().raw_tcp_addr().unwrap();
+ #[allow(deprecated)]
+ let command = Command::cargo_bin("iggy-bench").unwrap();
+ let output = assert_cmd::Command::from_std(command)
+ .args([
+ "--total-data",
+ REPORT_TEST_DATA,
+ "pinned-producer",
+ "--producers",
+ "1",
+ "--streams",
+ "1",
+ "tcp",
+ "--server-address",
+ &server_address,
+ "output",
+ "--output-dir",
+ ])
+ .arg(output_dir.path())
+ .envs([
+ ("IGGY_ROOT_PASSWORD", TEST_SECRET),
+ ("IGGY_CLUSTER_AUTH_SHARED_SECRET", TEST_SECRET),
+ ("IGGY_HTTP_JWT_ENCODING_SECRET", TEST_SECRET),
+ ("IGGY_ENCRYPTION_KEY", TEST_SECRET),
+ ("IGGY_UNRELATED_CREDENTIAL", TEST_SECRET),
+ ("IGGY_CONFIG_PATH", "benchmark-config.toml"),
+ ("IGGY_ENV_PATH", "benchmark.env"),
+ ("IGGY_SHARDING_CPU_ALLOCATION", "4"),
+ ("IGGY_SHARD_RUNTIME_CAPACITY", "8192"),
+ ("IGGY_SHARD_EVENT_INTERVAL", "256"),
+ ])
+ .timeout(BENCH_WAIT_TIMEOUT)
+ .unwrap();
+ let report_dir = fs::read_dir(output_dir.path())
+ .unwrap()
+ .next()
+ .unwrap()
+ .unwrap()
+ .path();
+ let report = fs::read_to_string(report_dir.join("report.json")).unwrap();
+ assert!(report.contains("IGGY_CONFIG_PATH=benchmark-config.toml"));
+ assert!(report.contains("IGGY_ENV_PATH=benchmark.env"));
+ assert!(report.contains("IGGY_SHARDING_CPU_ALLOCATION=4"));
+ assert!(report.contains("IGGY_SHARD_RUNTIME_CAPACITY=8192"));
+ assert!(report.contains("IGGY_SHARD_EVENT_INTERVAL=256"));
+ for (source, content) in [
+ ("report", report.as_str()),
+ ("stdout", &String::from_utf8_lossy(&output.stdout)),
+ ("stderr", &String::from_utf8_lossy(&output.stderr)),
+ ] {
+ assert!(
+ !content.contains(TEST_SECRET),
+ "{source} exposes a credential"
+ );
+ }
+}
diff --git
a/core/integration/tests/cli/consumer_group/test_consumer_group_create_command.rs
b/core/integration/tests/cli/consumer_group/test_consumer_group_create_command.rs
index f1400d2cc..856d7dcb6 100644
---
a/core/integration/tests/cli/consumer_group/test_consumer_group_create_command.rs
+++
b/core/integration/tests/cli/consumer_group/test_consumer_group_create_command.rs
@@ -242,11 +242,11 @@ pub async fn should_help_match() {
.execute_test_for_help_command(TestHelpCmd::new(
vec!["consumer-group", "create", "--help"],
format!(
- r#"Create consumer group with given ID and name for given
stream ID and topic ID.
+ r#"Create consumer group with given name for given stream ID
and topic ID.
Stream ID can be specified as a stream name or ID
Topic ID can be specified as a topic name or ID
-If group ID is not provided then the server will automatically assign it
+The server assigns the group ID. The legacy --group-id flag is ignored.
Examples:
iggy consumer-group create 1 1 prod
@@ -272,7 +272,7 @@ Arguments:
Options:
-g, --group-id <GROUP_ID>
- Consumer group ID to create
+ Legacy consumer group ID flag (ignored)
-h, --help
Print help (see a summary with '-h')
@@ -291,7 +291,7 @@ pub async fn should_short_help_match() {
.execute_test_for_help_command(TestHelpCmd::new(
vec!["consumer-group", "create", "-h"],
format!(
- r#"Create consumer group with given ID and name for given
stream ID and topic ID.
+ r#"Create consumer group with given name for given stream ID
and topic ID.
{USAGE_PREFIX} consumer-group create [OPTIONS] <STREAM_ID> <TOPIC_ID> <NAME>
@@ -301,7 +301,7 @@ Arguments:
<NAME> Consumer group name to create
Options:
- -g, --group-id <GROUP_ID> Consumer group ID to create
+ -g, --group-id <GROUP_ID> Legacy consumer group ID flag (ignored)
-h, --help Print help (see more with '--help')
"#,
),
diff --git
a/core/integration/tests/cli/consumer_group/test_consumer_group_help_command.rs
b/core/integration/tests/cli/consumer_group/test_consumer_group_help_command.rs
index de3dfd977..0b7e12b05 100644
---
a/core/integration/tests/cli/consumer_group/test_consumer_group_help_command.rs
+++
b/core/integration/tests/cli/consumer_group/test_consumer_group_help_command.rs
@@ -32,7 +32,7 @@ pub async fn should_help_match() {
{USAGE_PREFIX} consumer-group <COMMAND>
Commands:
- create Create consumer group with given ID and name for given stream ID and
topic ID. [alias: c]
+ create Create consumer group with given name for given stream ID and topic
ID. [alias: c]
delete Delete consumer group with given ID for given stream ID and topic ID
[alias: d]
get Get details of a single consumer group with given ID for given
stream ID and topic ID [alias: g]
list List all consumer groups for given stream ID and topic ID [alias: l]
diff --git
a/core/integration/tests/cli/consumer_offset/test_consumer_offset_get_command.rs
b/core/integration/tests/cli/consumer_offset/test_consumer_offset_get_command.rs
index 41288c58b..948032539 100644
---
a/core/integration/tests/cli/consumer_offset/test_consumer_offset_get_command.rs
+++
b/core/integration/tests/cli/consumer_offset/test_consumer_offset_get_command.rs
@@ -277,7 +277,7 @@ Arguments:
Options:
-k, --kind <KIND>
- Consumer kind: "consumer" for regular consumer, "consumer_group" for
consumer group
+ Consumer kind: "consumer" for regular consumer, "consumer-group" for
consumer group
Possible values:
- consumer: `Consumer` represents a regular consumer
@@ -313,7 +313,7 @@ Arguments:
<PARTITION_ID> Partitions ID for which consumer offset is retrieved
Options:
- -k, --kind <KIND> Consumer kind: "consumer" for regular consumer,
"consumer_group" for consumer group [default: consumer] [possible values:
consumer, consumer-group]
+ -k, --kind <KIND> Consumer kind: "consumer" for regular consumer,
"consumer-group" for consumer group [default: consumer] [possible values:
consumer, consumer-group]
-h, --help Print help (see more with '--help')
"#,
),
diff --git
a/core/integration/tests/cli/consumer_offset/test_consumer_offset_set_command.rs
b/core/integration/tests/cli/consumer_offset/test_consumer_offset_set_command.rs
index 1de493808..9b8cd5c43 100644
---
a/core/integration/tests/cli/consumer_offset/test_consumer_offset_set_command.rs
+++
b/core/integration/tests/cli/consumer_offset/test_consumer_offset_set_command.rs
@@ -335,7 +335,7 @@ Arguments:
Options:
-k, --kind <KIND>
- Consumer kind: "consumer" for regular consumer, "consumer_group" for
consumer group
+ Consumer kind: "consumer" for regular consumer, "consumer-group" for
consumer group
Possible values:
- consumer: `Consumer` represents a regular consumer
@@ -372,7 +372,7 @@ Arguments:
<OFFSET> Offset to set
Options:
- -k, --kind <KIND> Consumer kind: "consumer" for regular consumer,
"consumer_group" for consumer group [default: consumer] [possible values:
consumer, consumer-group]
+ -k, --kind <KIND> Consumer kind: "consumer" for regular consumer,
"consumer-group" for consumer group [default: consumer] [possible values:
consumer, consumer-group]
-h, --help Print help (see more with '--help')
"#,
),
diff --git a/core/integration/tests/cli/general/test_help_command.rs
b/core/integration/tests/cli/general/test_help_command.rs
index a7ea9e2a0..953af856b 100644
--- a/core/integration/tests/cli/general/test_help_command.rs
+++ b/core/integration/tests/cli/general/test_help_command.rs
@@ -151,12 +151,12 @@ Options:
--quic-send-window <QUIC_SEND_WINDOW>
The optional send window for QUIC
{CLAP_INDENT}
- [default: 100000]
+ [default: 1000000]
--quic-receive-window <QUIC_RECEIVE_WINDOW>
The optional receive window for QUIC
{CLAP_INDENT}
- [default: 100000]
+ [default: 1000000]
--quic-response-buffer-size <QUIC_RESPONSE_BUFFER_SIZE>
The optional response buffer size for QUIC
@@ -171,7 +171,7 @@ Options:
--quic-max-idle-timeout <QUIC_MAX_IDLE_TIMEOUT>
The optional maximum idle timeout for QUIC
{CLAP_INDENT}
- [default: 10000]
+ [default: 100000]
--quic-validate-certificate
Flag to enable certificate validation for QUIC
@@ -192,7 +192,7 @@ Options:
[default: "1s"]
-q, --quiet
- Quiet mode (disabled stdout printing)
+ Suppress logging output (command-specific output may remain)
-d, --debug <DEBUG>
Debug mode (verbose printing to given file)
diff --git a/core/integration/tests/cli/message/test_message_flush_command.rs
b/core/integration/tests/cli/message/test_message_flush_command.rs
index c07b18968..d1501243a 100644
--- a/core/integration/tests/cli/message/test_message_flush_command.rs
+++ b/core/integration/tests/cli/message/test_message_flush_command.rs
@@ -196,12 +196,11 @@ pub async fn should_help_match() {
.execute_test_for_help_command(TestHelpCmd::new(
vec!["message", "flush", "--help"],
format!(
- r#"Flush messages from given topic ID and given stream ID
+ r#"Legacy message flush command (unsupported by the server)
-Command is used to force a flush of unsaved_buffer to disk
-for specific stream, topic and partition. If fsync is enabled
-then the data is flushed to disk and fsynced, otherwise the
-data is only flushed to disk.
+Binary transports return FeatureUnavailable. HTTP has no flush route.
+For acknowledgements backed by stable storage, create the topic with
+--durability persisted.
Stream ID can be specified as a stream name or ID
Topic ID can be specified as a topic name or ID
@@ -216,24 +215,23 @@ Examples:
Arguments:
<STREAM_ID>
- ID of the stream for which messages will be flushed
+ Stream ID for the flush request
{CLAP_INDENT}
Stream ID can be specified as a stream name or ID
<TOPIC_ID>
- ID of the topic for which messages will be flushed
+ Topic ID for the flush request
{CLAP_INDENT}
Topic ID can be specified as a topic name or ID
<PARTITION_ID>
- Partition ID for which messages will be flushed
+ Partition ID for the flush request
Options:
-f, --fsync
- fsync flushed data to disk
+ Request fsync (unsupported by the server)
{CLAP_INDENT}
- If option is enabled then the data is flushed to disk and fsynced,
- otherwise the data is only flushed to disk. Default is false.
+ The server rejects flush requests regardless of this flag.
-h, --help
Print help (see a summary with '-h')
@@ -252,17 +250,17 @@ pub async fn should_short_help_match() {
.execute_test_for_help_command(TestHelpCmd::new(
vec!["message", "flush", "-h"],
format!(
- r#"Flush messages from given topic ID and given stream ID
+ r#"Legacy message flush command (unsupported by the server)
{USAGE_PREFIX} message flush [OPTIONS] <STREAM_ID> <TOPIC_ID> <PARTITION_ID>
Arguments:
- <STREAM_ID> ID of the stream for which messages will be flushed
- <TOPIC_ID> ID of the topic for which messages will be flushed
- <PARTITION_ID> Partition ID for which messages will be flushed
+ <STREAM_ID> Stream ID for the flush request
+ <TOPIC_ID> Topic ID for the flush request
+ <PARTITION_ID> Partition ID for the flush request
Options:
- -f, --fsync fsync flushed data to disk
+ -f, --fsync Request fsync (unsupported by the server)
-h, --help Print help (see more with '--help')
"#,
),
diff --git a/core/integration/tests/cli/message/test_message_help_command.rs
b/core/integration/tests/cli/message/test_message_help_command.rs
index 8ef97c460..83e83d010 100644
--- a/core/integration/tests/cli/message/test_message_help_command.rs
+++ b/core/integration/tests/cli/message/test_message_help_command.rs
@@ -34,7 +34,7 @@ pub async fn should_help_match() {
Commands:
send Send messages to given topic ID and given stream ID [alias: s]
poll Poll messages from given topic ID and given stream ID [alias: p]
- flush Flush messages from given topic ID and given stream ID [alias: f]
+ flush Legacy message flush command (unsupported by the server) [alias: f]
help Print this message or the help of the given subcommand(s)
Options:
diff --git a/core/integration/tests/cli/message/test_message_send_command.rs
b/core/integration/tests/cli/message/test_message_send_command.rs
index a24a6c526..abde36aef 100644
--- a/core/integration/tests/cli/message/test_message_send_command.rs
+++ b/core/integration/tests/cli/message/test_message_send_command.rs
@@ -20,11 +20,13 @@ use crate::cli::common::{
};
use assert_cmd::assert::Assert;
use async_trait::async_trait;
+use iggy::prelude::defaults::{DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME};
use iggy::prelude::*;
-use predicates::str::diff;
+use predicates::str::{contains, diff};
use serial_test::parallel;
use std::collections::BTreeMap;
use std::str::from_utf8;
+use std::time::Duration;
use twox_hash::XxHash32;
#[derive(Debug)]
@@ -405,7 +407,7 @@ Options:
-m, --message-key <MESSAGE_KEY>
Messages key which will be used to partition the messages
{CLAP_INDENT}
- Value of the key will be used by the server to calculate the
partition ID
+ The key must contain 1 to 255 bytes. Binary clients resolve the
partition ID; HTTP resolves it on the server.
-H, --headers <HEADERS>
Comma separated list of key:kind:value, sent as header with the
message
@@ -461,3 +463,41 @@ Options:
))
.await;
}
+
+#[test]
+#[parallel]
+fn given_invalid_message_key_when_sending_should_reject_before_connecting() {
+ const UNREACHABLE_SERVER_ADDRESS: &str = "127.0.0.1:0";
+ const ARGUMENT_PARSE_TIMEOUT: Duration = Duration::from_secs(5);
+
+ let cli_home = tempfile::tempdir().unwrap();
+ let oversized_key = "x".repeat(usize::from(u8::MAX) + 1);
+ let oversized_unicode_key = "é".repeat(usize::from(u8::MAX) / "é".len() +
1);
+
+ for key in ["", oversized_key.as_str(), oversized_unicode_key.as_str()] {
+ #[allow(deprecated)]
+ let mut command = assert_cmd::Command::cargo_bin("iggy").unwrap();
+ command
+ .env("IGGY_HOME", cli_home.path())
+ .args([
+ "--tcp-server-address",
+ UNREACHABLE_SERVER_ADDRESS,
+ "-u",
+ DEFAULT_ROOT_USERNAME,
+ "-p",
+ DEFAULT_ROOT_PASSWORD,
+ "message",
+ "send",
+ "--message-key",
+ key,
+ "stream",
+ "topic",
+ "payload",
+ ])
+ .timeout(ARGUMENT_PARSE_TIMEOUT)
+ .assert()
+ .code(2)
+ .stderr(contains("--message-key <MESSAGE_KEY>"))
+ .stderr(contains("Invalid command"));
+ }
+}
diff --git
a/core/integration/tests/cli/personal_access_token/test_pat_create_command.rs
b/core/integration/tests/cli/personal_access_token/test_pat_create_command.rs
index ec0ec77ad..613e42789 100644
---
a/core/integration/tests/cli/personal_access_token/test_pat_create_command.rs
+++
b/core/integration/tests/cli/personal_access_token/test_pat_create_command.rs
@@ -130,7 +130,7 @@ pub async fn should_help_match() {
Create personal access token which allow authenticating the clients using
a token, instead of the regular credentials (username and password)
-In quiet mode only the personal access token name is printed
+In quiet mode the raw token is printed unless --store-token is set
Examples
iggy pat create name
@@ -153,7 +153,7 @@ Options:
Store token in an underlying platform-specific secure store
{CLAP_INDENT}
Generated token is stored in a platform-specific secure storage
without revealing its content to the user. It can be used to authenticate on
iggy server using associated name and
- -n/--token-name command line option instead of -u/--username and
-p/--password or -t/--token. In quiet mode only the token name is printed. This
option can only be used for creating tokens
+ -n/--token-name command line option instead of -u/--username and
-p/--password or -t/--token. Quiet mode prints no token or confirmation. This
option can only be used for creating tokens
which does not have expiry time set.
-h, --help
diff --git a/core/integration/tests/cli/stream/test_stream_create_command.rs
b/core/integration/tests/cli/stream/test_stream_create_command.rs
index 78592afc2..19d6f7436 100644
--- a/core/integration/tests/cli/stream/test_stream_create_command.rs
+++ b/core/integration/tests/cli/stream/test_stream_create_command.rs
@@ -114,7 +114,7 @@ pub async fn should_help_match() {
format!(
r#"Create stream with given name
-If stream ID is not provided then the server will automatically assign it
+The server assigns the stream ID. The legacy --stream-id flag is ignored.
Examples:
iggy stream create prod
@@ -128,7 +128,7 @@ Arguments:
Options:
-s, --stream-id <STREAM_ID>
- Stream ID to create
+ Legacy stream ID flag (ignored)
-h, --help
Print help (see a summary with '-h')
@@ -155,7 +155,7 @@ Arguments:
<NAME> Name of the stream
Options:
- -s, --stream-id <STREAM_ID> Stream ID to create
+ -s, --stream-id <STREAM_ID> Legacy stream ID flag (ignored)
-h, --help Print help (see more with '--help')
"#,
),
diff --git a/core/integration/tests/cli/system/test_me_command.rs
b/core/integration/tests/cli/system/test_me_command.rs
index cb67e457e..cb92c9cee 100644
--- a/core/integration/tests/cli/system/test_me_command.rs
+++ b/core/integration/tests/cli/system/test_me_command.rs
@@ -195,7 +195,7 @@ pub async fn should_help_match() {
r#"get current client info
Command connects to Iggy server and collects client info like client ID, user
ID
-server address and protocol type.
+client address as seen by the server and protocol type.
{USAGE_PREFIX} me
diff --git a/core/integration/tests/cli/system/test_snapshot_cmd.rs
b/core/integration/tests/cli/system/test_snapshot_cmd.rs
index da489853a..2aab0b264 100644
--- a/core/integration/tests/cli/system/test_snapshot_cmd.rs
+++ b/core/integration/tests/cli/system/test_snapshot_cmd.rs
@@ -138,7 +138,7 @@ Options:
{CLAP_INDENT}
Examples:
- `--compression bzip2` for higher compression.
- - `--compression none` to store without compression.
+ - `--compression stored` to store without compression.
-s, --snapshot-types <SNAPSHOT_TYPES>...
Specify types of snapshots to include.
diff --git a/core/integration/tests/cli/topic/test_topic_create_command.rs
b/core/integration/tests/cli/topic/test_topic_create_command.rs
index de4f8e364..89a7be9e2 100644
--- a/core/integration/tests/cli/topic/test_topic_create_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_create_command.rs
@@ -243,7 +243,7 @@ pub async fn should_help_match() {
r#"Create topic with given name, number of partitions,
compression algorithm and expiry time for given stream ID
Stream ID can be specified as a stream name or ID
-If topic ID is not provided then the server will automatically assign it
+The server assigns the topic ID. The legacy --topic-id flag is ignored.
Examples
iggy topic create 1 sensor1 2 gzip 15days
@@ -266,24 +266,24 @@ Arguments:
Number of partitions inside the topic
<COMPRESSION_ALGORITHM>
- Compression algorithm for the topic, set to "none" for no compression
+ Compression metadata (none or gzip). Payload compression is not
implemented
[MESSAGE_EXPIRY]...
Message expiry time in human-readable format like "unlimited" or
"15days 2min 2s"
{CLAP_INDENT}
- "server_default" or skipping parameter makes CLI to use server
default (from current server config) expiry time
+ Skipping this parameter or using "server_default" creates a topic
with no message expiry.
{CLAP_INDENT}
[default: server_default]
Options:
-t, --topic-id <TOPIC_ID>
- Topic ID to create
+ Legacy topic ID flag (ignored)
-m, --max-topic-size <MAX_TOPIC_SIZE>
Max topic size in human-readable format like "unlimited" or "15GB"
{CLAP_INDENT}
- "server_default" or skipping parameter makes CLI to use server
default (from current server config) max topic size
- Can't be lower than segment size in the config.
+ Skipping this parameter or using "server_default" creates a topic
with unlimited size.
+ A finite size cannot be lower than the topic segment size.
{CLAP_INDENT}
[default: server_default]
@@ -336,11 +336,11 @@ Arguments:
<STREAM_ID> Stream ID to create topic
<NAME> Name of the topic
<PARTITIONS_COUNT> Number of partitions inside the topic
- <COMPRESSION_ALGORITHM> Compression algorithm for the topic, set to "none"
for no compression
+ <COMPRESSION_ALGORITHM> Compression metadata (none or gzip). Payload
compression is not implemented
[MESSAGE_EXPIRY]... Message expiry time in human-readable format like
"unlimited" or "15days 2min 2s" [default: server_default]
Options:
- -t, --topic-id <TOPIC_ID> Topic ID to
create
+ -t, --topic-id <TOPIC_ID> Legacy topic
ID flag (ignored)
-m, --max-topic-size <MAX_TOPIC_SIZE> Max topic
size in human-readable format like "unlimited" or "15GB" [default:
server_default]
--durability <DURABILITY> Message
completion policy: replicated or persisted. Both policies store messages on
disk [default: replicated] [possible values:
replicated,
persisted]
diff --git a/core/integration/tests/cli/topic/test_topic_update_command.rs
b/core/integration/tests/cli/topic/test_topic_update_command.rs
index 9cad22689..d284ab2a6 100644
--- a/core/integration/tests/cli/topic/test_topic_update_command.rs
+++ b/core/integration/tests/cli/topic/test_topic_update_command.rs
@@ -286,11 +286,11 @@ Stream ID can be specified as a stream name or ID
Topic ID can be specified as a topic name or ID
Examples
- iggy update 1 1 sensor3 none
- iggy update prod sensor3 old-sensor none
- iggy update test debugs ready gzip 15days
- iggy update 1 1 new-name gzip
- iggy update 1 2 new-name none 1day 1hour 1min 1sec
+ iggy topic update 1 1 sensor3 none
+ iggy topic update prod sensor3 old-sensor none
+ iggy topic update test debugs ready gzip 15days
+ iggy topic update 1 1 new-name gzip
+ iggy topic update 1 2 new-name none 1day 1hour 1min 1sec
{USAGE_PREFIX} topic update [OPTIONS] <STREAM_ID> <TOPIC_ID> <NAME>
<COMPRESSION_ALGORITHM> [MESSAGE_EXPIRY]...
@@ -309,12 +309,12 @@ Arguments:
New name for the topic
<COMPRESSION_ALGORITHM>
- Compression algorithm for the topic, set to "none" for no compression
+ Compression metadata (none or gzip). Payload compression is not
implemented
[MESSAGE_EXPIRY]...
New message expiry time in human-readable format like "unlimited" or
"15days 2min 2s"
{CLAP_INDENT}
- "server_default" or skipping parameter makes CLI to use server
default (from current server config) expiry time
+ Skipping this parameter or using "server_default" preserves the
current message expiry.
{CLAP_INDENT}
[default: server_default]
@@ -322,8 +322,8 @@ Options:
-m, --max-topic-size <MAX_TOPIC_SIZE>
New max topic size in human-readable format like "unlimited" or
"15GB"
{CLAP_INDENT}
- "server_default" or skipping parameter makes CLI to use server
default (from current server config) max topic size
- Can't be lower than segment size in the config.
+ Skipping this parameter or using "server_default" preserves the
current max topic size.
+ A finite size cannot be lower than the topic segment size.
{CLAP_INDENT}
[default: server_default]
@@ -352,7 +352,7 @@ Arguments:
<STREAM_ID> Stream ID to update topic
<TOPIC_ID> Topic ID to update
<NAME> New name for the topic
- <COMPRESSION_ALGORITHM> Compression algorithm for the topic, set to "none"
for no compression
+ <COMPRESSION_ALGORITHM> Compression metadata (none or gzip). Payload
compression is not implemented
[MESSAGE_EXPIRY]... New message expiry time in human-readable format
like "unlimited" or "15days 2min 2s" [default: server_default]
Options:
diff --git a/core/integration/tests/cluster/fast_primary_rejoin.rs
b/core/integration/tests/cluster/fast_primary_rejoin.rs
index 6856654c7..90d69c228 100644
--- a/core/integration/tests/cluster/fast_primary_rejoin.rs
+++ b/core/integration/tests/cluster/fast_primary_rejoin.rs
@@ -42,6 +42,7 @@ use crate::server::http_client::HttpClient;
const STREAM_NAME: &str = "rejoin-stream";
const TOPIC_NAME: &str = "rejoin-topic";
const PARTITION_ID: u32 = 0;
+const DURABILITY_HEADER: &str = "iggy-durability";
/// Acks the pinned producer must capture before any disruption, so the
/// session is warm and mid-stream rather than freshly connected.
@@ -435,6 +436,14 @@ async fn
given_http_writes_on_a_rejoined_backup_when_the_primary_moved_should_fo
.await
.expect("forwarded HTTP produce");
assert_eq!(response.status(), StatusCode::CREATED);
+ assert_eq!(
+ response
+ .headers()
+ .get(DURABILITY_HEADER)
+ .and_then(|value| value.to_str().ok()),
+ Some(<&str>::from(Durability::Persisted)),
+ "forwarded produce must preserve the primary's durability guarantee"
+ );
let confirmation: SendMessagesConfirmations =
response.json().await.expect("decode confirmations");
let confirmation = confirmation
diff --git a/core/integration/tests/mod.rs b/core/integration/tests/mod.rs
index 7bdcd5235..d34f2b00c 100644
--- a/core/integration/tests/mod.rs
+++ b/core/integration/tests/mod.rs
@@ -30,6 +30,7 @@ use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;
use tracing_subscriber::{EnvFilter, fmt};
+mod bench;
// Drives the `iggy` CLI binary against a running server. Single-node and the
// default 3-node cluster both pass.
mod cli;
diff --git a/core/integration/tests/server/mod.rs
b/core/integration/tests/server/mod.rs
index 85a690696..159b9217b 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -89,3 +89,4 @@ mod message_retrieval;
mod purge_delete;
mod scenarios;
mod specific;
+mod telemetry;
diff --git a/core/integration/tests/server/telemetry.rs
b/core/integration/tests/server/telemetry.rs
new file mode 100644
index 000000000..1306aaa39
--- /dev/null
+++ b/core/integration/tests/server/telemetry.rs
@@ -0,0 +1,63 @@
+// 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 iggy::prelude::*;
+use integration::harness::TestHarness;
+use wiremock::{Mock, MockServer, ResponseTemplate, matchers::method};
+
+#[tokio::test]
+#[serial_test::parallel]
+async fn given_http_telemetry_enabled_when_starting_should_serve_requests() {
+ let collector = MockServer::start().await;
+ Mock::given(method("POST"))
+ .respond_with(ResponseTemplate::new(200))
+ .mount(&collector)
+ .await;
+ let mut harness = TestHarness::builder()
+ .default_server()
+ .cluster_nodes(1)
+ .build()
+ .expect("build telemetry harness");
+ for (key, value) in [
+ ("IGGY_TELEMETRY_ENABLED", "true".to_string()),
+ ("IGGY_TELEMETRY_LOGS_TRANSPORT", "http".to_string()),
+ ("IGGY_TELEMETRY_TRACES_TRANSPORT", "http".to_string()),
+ (
+ "IGGY_TELEMETRY_LOGS_ENDPOINT",
+ format!("{}/v1/logs", collector.uri()),
+ ),
+ (
+ "IGGY_TELEMETRY_TRACES_ENDPOINT",
+ format!("{}/v1/traces", collector.uri()),
+ ),
+ ] {
+ harness.server_mut().add_env(key, value);
+ }
+ harness
+ .start()
+ .await
+ .expect("HTTP telemetry must not prevent server startup");
+ let client = harness
+ .server()
+ .tcp_client()
+ .expect("TCP client")
+ .with_root_login()
+ .connect()
+ .await
+ .expect("connect with HTTP telemetry enabled");
+ client.ping().await.expect("server must answer ping");
+}
diff --git a/core/integration/tests/server/topic_admission_vsr.rs
b/core/integration/tests/server/topic_admission_vsr.rs
index 74e51813a..c4a14580e 100644
--- a/core/integration/tests/server/topic_admission_vsr.rs
+++ b/core/integration/tests/server/topic_admission_vsr.rs
@@ -22,10 +22,10 @@
//! `max_topic_size` below the configured segment size denies with
//! `InvalidTopicSize`; `ServerDefault` and `Unlimited` sizes pass. Update
//! enforces the same size floor, since a topic capped below one segment can
-//! never rotate however it acquired that cap. Update otherwise stores
-//! `max_topic_size` and `message_expiry` verbatim and gets echo the stored
-//! value (never the node default frozen at update time), matching legacy wire
-//! behavior. Deleting more partitions than the topic has rejects
+//! never rotate however it acquired that cap. Updates preserve existing
+//! `max_topic_size` and `message_expiry` values for omitted keys and default
+//! sentinels. Explicit values are echoed in both typed fields and options.
+//! Deleting more partitions than the topic has rejects
//! with `InvalidPartitionsCount` as a committed result instead of silently
//! acking a no-op. Listing topics of a missing stream replies with an empty
//! list, as the legacy server does.
@@ -251,20 +251,20 @@ async fn
given_updated_topic_when_getting_topic_should_echo_stored_values(harnes
.await
.expect("get topic")
.expect("topic exists");
+ let reported_options =
TopicCreateOptions::from_resource_options(&topic.options);
+ assert_eq!(reported_options.max_topic_size,
Some(topic.max_topic_size));
+ assert_eq!(reported_options.message_expiry,
Some(topic.message_expiry));
(topic.max_topic_size, topic.message_expiry)
}
};
- // Settings ride the options block and 0 is its "resolve the default"
- // sentinel, so a `ServerDefault` on update carries no key at all: the
topic
- // keeps what it already had. Resetting a setting back to the node default
- // is deliberately not expressible -- an update states the values it wants,
- // and everything it omits survives.
+ // ServerDefault is a no-op on update, so it must preserve the effective
+ // value in both the typed fields and the reported options.
let created_size =
MaxTopicSize::Custom(IggyByteSize::from_str("2GiB").expect("byte size"));
assert_eq!(
update_topic(MaxTopicSize::ServerDefault,
IggyExpiry::ServerDefault).await,
(created_size, IggyExpiry::NeverExpire),
- "a sentinel carries no key, so the value set at creation survives"
+ "a sentinel leaves the value set at creation unchanged"
);
let custom_size =
MaxTopicSize::Custom(IggyByteSize::from_str("3GiB").expect("byte size"));
let custom_expiry =
IggyExpiry::ExpireDuration(IggyDuration::from_str("5s").expect("duration"));
diff --git a/core/metadata/src/stm/snapshot.rs
b/core/metadata/src/stm/snapshot.rs
index 2185b439e..423e280f7 100644
--- a/core/metadata/src/stm/snapshot.rs
+++ b/core/metadata/src/stm/snapshot.rs
@@ -22,24 +22,16 @@ use std::fmt;
use crate::stm::stream::StreamsSnapshot;
use crate::stm::user::UsersSnapshot;
-/// The version of the snapshot format in use, reserved for breaking changes.
+/// The snapshot format version written by this build.
///
-/// One version means exactly one serialized shape, and
[`MetadataSnapshot::decode`]
-/// accepts nothing else. Bump it in the same change that alters the shape:
append,
-/// remove, reorder, retype, or redefine the meaning of any field, at any depth
-/// under [`MetadataSnapshot`]. There is no accepted range and no per-version
-/// translation. A snapshot this build cannot read is refused, not best-effort
-/// decoded.
+/// Each version identifies a serialized shape. Bump it when fields change,
+/// including nested fields: `MessagePack` encodes structs positionally, so a
+/// layout change can reinterpret bytes without a deserialization error.
///
-/// Nothing softer would hold. msgpack encodes a struct positionally, so a
field one
-/// build appends reaches another as an unexplained extra array element;
without the
-/// version the reader either fails on an unrelated msgpack error or, for a
-/// same-length change, silently reads one field's bytes as another's.
-///
-/// Bumping it invalidates every `snapshot.bin` already on disk, and a node
whose
-/// snapshot is refused refuses boot. That is deliberate. The metadata plane is
-/// pre-production, so the cost is clearing a data directory; once it ships, a
bump
-/// needs an explicit translation path added here alongside it.
+/// [`MetadataSnapshot::decode`] accepts versions from
+/// [`MIN_READABLE_SNAPSHOT_FORMAT_VERSION`] through this version. Older
accepted
+/// formats rely on defaulted trailing fields; incompatible versions outside
+/// that range are refused before payload deserialization.
///
/// Version 2: `status` sits at reply-header offset 216 (version 1 carried a
/// `namespace` word before it), which the client table's cached replies embed
as raw
@@ -283,9 +275,9 @@ impl MetadataSnapshot {
/// whichever later field happened to misparse.
///
/// # Errors
- /// [`SnapshotError::UnsupportedFormatVersion`] when the stamped version
is not
- /// [`SNAPSHOT_FORMAT_VERSION`], or [`SnapshotError::Deserialize`] if
msgpack
- /// deserialization fails.
+ /// [`SnapshotError::UnsupportedFormatVersion`] when the stamped version is
+ /// outside
[`MIN_READABLE_SNAPSHOT_FORMAT_VERSION`]..=[`SNAPSHOT_FORMAT_VERSION`],
+ /// or [`SnapshotError::Deserialize`] if `MessagePack` deserialization
fails.
pub fn decode(bytes: &[u8]) -> Result<Self, SnapshotError> {
// Bytes carrying no readable version are not a snapshot at all, so
they fall
// through to the deserializer, whose error names what actually went
wrong.
diff --git a/core/metadata/src/stm/stream.rs b/core/metadata/src/stm/stream.rs
index 9844d552d..ae03bade5 100644
--- a/core/metadata/src/stm/stream.rs
+++ b/core/metadata/src/stm/stream.rs
@@ -67,6 +67,7 @@ use
iggy_common::wire_conversions::{resource_options_from_wire, resource_options
use iggy_common::{
CompressionAlgorithm, IggyByteSize, IggyExpiry, IggyTimestamp,
MaxTopicSize, PartitionStats,
ResourceOptions, StreamStats, TopicCreateOptions, TopicRuntimeOptions,
TopicStats,
+ topic_option_keys,
};
use serde::{Deserialize, Serialize};
use server_common::sharding::{IggyNamespace, MAX_PARTITIONS, MAX_STREAMS,
MAX_TOPICS};
@@ -2234,20 +2235,23 @@ impl StateHandler for UpdateTopicRequest {
// Decoded before any mutation: a malformed block must leave the topic
// untouched rather than half-renamed.
- let Ok(updated_options) = resource_options_from_wire(&self.options,
true) else {
+ let Ok(mut updated_options) =
resource_options_from_wire(&self.options, true) else {
return ApplyReply::err(UpdateTopicResult::InvalidOptionValue);
};
// Read leniently, like every other committed op: a key this build does
// not know is skipped rather than failing an operation its peers
// accepted.
let updated = TopicCreateOptions::parse_committed(&self.options);
+ // Default sentinels leave both the effective value and its provenance
+ // unchanged, just as an omitted key does.
+ updated_options.retain(|key, _| match key.as_str() {
+ Ok(topic_option_keys::MESSAGE_EXPIRY) =>
updated.message_expiry.is_some(),
+ Ok(topic_option_keys::MAX_TOPIC_SIZE) =>
updated.max_topic_size.is_some(),
+ _ => true,
+ });
stream.topic_index.remove(&topic.name);
topic.name = new_name_arc.clone();
- // Settings arrive only through the options block now, so the typed
- // fields are a projection of it and cannot drift. Absent means absent:
- // a client that sends just a rename leaves every setting alone, and
one
- // built before a key existed cannot erase it.
if let Some(compression_algorithm) = updated.compression_algorithm {
topic.compression_algorithm = compression_algorithm;
}
@@ -2751,7 +2755,7 @@ mod tests {
CreateTopicRequest as WireCreateTopicRequest,
CreateTopicWithAssignmentsRequest,
};
use iggy_binary_protocol::responses::topics::get_topic::GetTopicResponse;
- use iggy_common::{HeaderKey, HeaderKind, topic_option_keys};
+ use iggy_common::{HeaderKey, HeaderKind, TopicUpdateOptions,
topic_option_keys};
use std::str::FromStr;
#[test]
@@ -2956,6 +2960,89 @@ mod tests {
);
}
+ #[test]
+ fn update_topic_sentinels_preserve_effective_options_and_provenance() {
+ let mut inner = StreamsInner::new();
+ create_stream(&mut inner, "stream");
+ let message_expiry = IggyExpiry::from(5_000_000u64);
+ let max_topic_size = MaxTopicSize::from(10_000_000_000u64);
+ let create = CreateTopicWithAssignmentsRequest {
+ created_view: 0,
+ request: WireCreateTopicRequest {
+ stream_id: WireIdentifier::numeric(0),
+ partitions_count: 1,
+ name: WireName::new("topic").unwrap(),
+ options: TopicCreateOptions {
+ message_expiry: Some(message_expiry),
+ ..TopicCreateOptions::default()
+ }
+ .to_explicit_wire(|key| key ==
topic_option_keys::MESSAGE_EXPIRY)
+ .unwrap(),
+ },
+ derived_options: TopicCreateOptions {
+ max_topic_size: Some(max_topic_size),
+ ..TopicCreateOptions::default()
+ }
+ .to_wire()
+ .unwrap(),
+ partitions: vec![CreatedPartitionAssignment {
+ partition_id: 0,
+ consensus_group_id: 1,
+ }],
+ };
+ assert_eq!(
+ StateHandler::apply(&create, &mut inner,
IggyTimestamp::from(1)).code,
+ 0
+ );
+ let original_options = inner
+ .items
+ .get(0)
+ .unwrap()
+ .topics
+ .get(0)
+ .unwrap()
+ .options
+ .clone();
+
+ for options in [
+ TopicUpdateOptions::default(),
+ TopicUpdateOptions {
+ message_expiry: Some(IggyExpiry::ServerDefault),
+ max_topic_size: Some(MaxTopicSize::ServerDefault),
+ ..TopicUpdateOptions::default()
+ },
+ TopicUpdateOptions {
+ raw: [
+ topic_option_keys::MESSAGE_EXPIRY,
+ topic_option_keys::MAX_TOPIC_SIZE,
+ ]
+ .into_iter()
+ .map(|key| (key.to_owned(), "server_default".to_owned()))
+ .collect(),
+ ..TopicUpdateOptions::default()
+ },
+ ] {
+ let update = UpdateTopicRequest {
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(0),
+ name: WireName::new("renamed").unwrap(),
+ options: options.to_wire().unwrap(),
+ };
+ assert_eq!(
+ StateHandler::apply(&update, &mut inner,
IggyTimestamp::from(2)).code,
+ 0
+ );
+ let topic = inner.items.get(0).unwrap().topics.get(0).unwrap();
+ assert_eq!(topic.name.as_ref(), "renamed");
+ assert_eq!(topic.message_expiry, message_expiry);
+ assert_eq!(topic.max_topic_size, max_topic_size);
+ assert_eq!(
+ topic.options, original_options,
+ "update {options:?} must preserve values and provenance"
+ );
+ }
+ }
+
#[test]
fn create_stream_options_survive_snapshot_roundtrip() {
use crate::stm::snapshot::FillSnapshot;
diff --git a/core/sdk/src/vsr.rs b/core/sdk/src/vsr.rs
index 876457ad8..f8ceae42d 100644
--- a/core/sdk/src/vsr.rs
+++ b/core/sdk/src/vsr.rs
@@ -95,21 +95,17 @@ pub(crate) fn encode_request_header(
session.session().unwrap_or(0),
)
} else {
- // Partition ops consume an id too, even though nothing dedups
- // them yet: dedup needs each send to carry a distinct number,
- // and the metadata watermark tolerates the resulting gaps
+ // Partition dedup needs each new write to carry a distinct id.
+ // The metadata watermark tolerates the resulting gaps
// (`client_table.rs`: "There is no `RequestGap`").
let session_id =
session.session().ok_or(IggyError::Unauthenticated)?;
(operation, session.next_request_id(), session_id)
}
}
};
- // Stamped only for the ops the server's `ClientTable` dedups, and only by
this
- // SDK: the others leave the field zero, which the server reads as
unstamped.
- // Partition ops are the large payloads and already carry `batch_checksum`
over
- // the same bytes, and nothing dedups them, so hashing here would only buy
the
- // server a second full-payload pass in `verify_request_checksum`.
NonReplicated
- // ops bypass dedup too.
+ // Metadata dedup compares this stamp with its cached replies. Partition
+ // dedup tracks request ids without retaining replies or comparing stamps;
+ // send batches carry their own checksum. NonReplicated ops bypass dedup.
let request_checksum = if operation.is_partition() || operation ==
Operation::NonReplicated {
0
} else {
@@ -135,12 +131,8 @@ pub(crate) fn encode_request_header(
// predating this sends. A server that rewrites the body (PAT,
password)
// carries it through untouched, so it keeps describing what the
client sent.
request_checksum,
- // Zeroed: the field is "informational" -- the server copies it into
- // `ReplyHeader.timestamp` for RTT but nothing else reads it. Paying
- // a `clock_gettime` syscall per encoded request (formerly held the
- // `consensus_session` lock too) for an unused field is waste.
- // Reintroduce a real stamp here when an RTT consumer actually wires
- // it up.
+ // Replicated prepares get a server timestamp. Direct replies may echo
+ // this field, but no RTT consumer needs a client clock read here.
timestamp: 0,
reserved,
..Default::default()
@@ -212,7 +204,7 @@ pub(crate) fn decode_response(response: Bytes) ->
Result<Bytes, IggyError> {
}
/// Decode a reply when the header and body have been read into separate
-/// buffers. Saves the 64B header `put_slice` that `decode_response` would
+/// buffers. Saves the 256-byte header `put_slice` that `decode_response` would
/// otherwise perform when callers concatenate header + body before decoding.
///
/// Also surfaces session-terminal `Command::Eviction` frames as typed
@@ -273,21 +265,15 @@ fn read_operation(header_bytes: &[u8; HEADER_SIZE]) ->
Result<Operation, IggyErr
.map_err(|_| IggyError::InvalidCommand)
}
-/// Interpret the committed result section that leads a metadata reply body.
+/// Strip the result section from metadata, consumer-offset write, and
+/// non-empty Register replies. Success carries `count == 0` then the payload;
+/// a business or transient rejection carries an error code in a result entry.
+/// A result section can report a pre-commit rejection, so its presence alone
+/// does not prove commitment.
///
-/// Metadata ops ([`Operation::is_metadata`]) commit a result
-/// section ahead of their typed payload (encode mirror:
-/// `metadata::stm::result::ApplyReply::write_reply_body`): success carries
-/// `count == 0` then the payload; a committed business rejection carries one
-/// `{index, result}` entry and no payload. Strip the section on success and
map
-/// a nonzero committed code to its [`IggyError`] -- the result discriminants
-/// share the `IggyError` code space, so this is the same
[`IggyError::from_code`]
-/// mapping the legacy transport applies to a status word.
-///
-/// Reads, the partition data plane, and Register/Logout carry no result
section
-/// and pass through untouched. A metadata body that is not a well-formed
result
-/// section is corruption, never a silent success, so it maps to
`InvalidCommand`
-/// rather than risk a rejection decoding as `Ok`.
+/// Reads, SendMessages, and Logout pass through untouched. An empty Register
+/// body passes through to fail the typed login decode. A malformed result
+/// section maps to `InvalidCommand` rather than decoding a rejection as `Ok`.
fn split_metadata_result(operation: Operation, body: Bytes) -> Result<Bytes,
IggyError> {
// Register (login/register) replies are result-framed too, so a transient
// login decodes to `TransientNotCommitted` and the SDK replays it. The one
@@ -500,12 +486,9 @@ mod tests {
}
#[test]
- fn request_checksum_is_stamped_only_for_deduped_operations() {
- // The stamp exists to stop a reused `request` number matching a dedup
- // entry recorded for different bytes, so it is worth its hashing pass
only
- // where `ClientTable` dedups. Partition ops are the large payloads and
- // already carry `batch_checksum` over the same bytes; NonReplicated
ops
- // bypass dedup. Neither stamps.
+ fn request_checksum_is_stamped_for_metadata_but_not_partition_operations()
{
+ // Metadata dedup compares the stamp against cached replies. Partition
+ // dedup checks request ids without stamps; NonReplicated bypasses
dedup.
let mut session = ConsensusSession::with_client_id(42);
session.bind(99);
let payload = Bytes::from_static(b"payload");
diff --git a/core/server/config.toml b/core/server/config.toml
index 0b8cdb518..618ca6cad 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -135,15 +135,15 @@ clock_skew = "5 s"
not_before = "0 s"
# Secret key for encoding JWTs.
-# If left empty, a secure random secret will be generated on each server start.
-# In cluster mode a configured secret (identical on every node) makes bearers
-# valid cluster-wide and activates follower-to-primary HTTP forwarding; with
-# cluster.auth enabled the key is instead derived from the shared PSK. Without
-# either, tokens are node-local and forwarding stays disabled.
+# If only one secret is set, it supplies both keys. With both empty, use the
+# cluster PSK when enabled, otherwise a random secret per start. Explicit JWT
+# secrets take precedence over the PSK. Identical keys on every node make
bearers
+# valid cluster-wide and activate follower-to-primary HTTP forwarding. Without
+# configured JWT secrets or a cluster PSK, tokens are node-local and
forwarding stays disabled.
encoding_secret = ""
# Secret key for decoding JWTs.
-# If left empty, a secure random secret will be generated on each server start.
+# When empty, the same fallback rules as encoding_secret apply.
decoding_secret = ""
# Indicates if the secret key is base64 encoded.
@@ -152,7 +152,7 @@ decoding_secret = ""
use_base64_secret = false
# Trusted issuers for A2A (Application-to-Application) authentication. Opt-in:
-# with none configured the listener accepts only self-issued HS256 tokens.
+# with none configured the listener accepts only self-issued HMAC tokens.
# `issuer`, `audience` and `jwks_url` are required per entry; `user_id` is
# optional but defaults to 0 (root), which is rejected - set it to the non-zero
# iggy user every token from that issuer is remapped onto.
@@ -397,7 +397,7 @@ retention = "7 days"
# Encryption configuration
[encryption]
-# Determines whether server-side data encryption for the messages payloads and
state commands is enabled (boolean).
+# Encrypt message payloads and user headers. Metadata and structural headers
remain unencrypted.
# `true` enables encryption for stored data using AES-256-GCM.
# `false` means data is stored without encryption.
enabled = false
@@ -416,14 +416,14 @@ enabled = true
# Size of the memory pool (string).
# Example: "512 MiB" or "1 GiB".
-# This defines the maximum, total memory allocated for the memory pool.
+# Buffers are allocated on demand; allocations beyond this budget use memory
outside the pool.
# Note: This number has to be multiplication of 4096 (default linux page size).
# Minimum size is 512 MiB due to internal implementation details.
size = "4 GiB"
# Maximum number of buffers in each bucket (u32).
-# There are 32 buckets in the memory pool. Each bucket can hold up to this
number of buffers
-# and holds different buffer sizes, from 256 B to 512 MiB.
+# There are 28 buckets in the memory pool. Each bucket can hold up to this
number of buffers
+# and holds different buffer sizes, from 4 KiB to 512 MiB.
# Note: This number has to be a power of 2. Minimum value is 128 due to
internal implementation details.
bucket_capacity = 8192
@@ -703,11 +703,11 @@ ports = { tcp = 8091, quic = 8081, http = 3001, websocket
= 8093, tcp_replica =
[sharding]
# CPU allocation - controls the number of shards and their CPU affinity.
# Possible values:
-# - "all": Use all available CPU cores (default)
-# - numeric value (e.g. 4): Use 4 shards (4 threads pinned to cores 0, 1, 2, 3)
+# - "all": Use all available CPU cores
+# - numeric value (e.g. 4): Use 4 shards on the first 4 CPUs in the allowed set
# - range (e.g. "5..8"): Use 3 shards with affinity to cores 5, 6, 7
# - numa settings:
-# + "numa:auto": Use all available numa node, cores
+# + "numa:auto": Use all available NUMA nodes and cores (default)
# + "numa:nodes=0,1;cores=4;no_ht=true": Use NUMA node 0 and 1, each nodes
use 4 cores, and no hyperthreads
cpu_allocation = "numa:auto"
@@ -925,7 +925,7 @@ validate_checksum = true
# Depth of a partition's prepare queue: how many uncommitted produce /
# consumer-offset ops may be in flight at once for that partition. Submits past
# it spill into a request queue of twice this depth; once both are full the
-# server drops the request without a reply and the client retries on its own
+# server replies with TransientNotAccepted. A lost reply can still cause a
client
# request timeout. Must be > 0 and <= 127: the ceiling is the view-change
wire, not
# memory. A DoViewChange describes the uncommitted suffix with one bit per op
in a
# u128 bitset, and this depth bounds that suffix.
diff --git a/core/server/src/args.rs b/core/server/src/args.rs
index 0317bfd96..9025c98f9 100644
--- a/core/server/src/args.rs
+++ b/core/server/src/args.rs
@@ -49,8 +49,8 @@ CONFIGURATION:
ENVIRONMENT VARIABLES:
Any configuration value can be overridden with an IGGY_ prefixed variable;
underscores separate the nested keys (IGGY_TCP_ADDRESS sets [tcp] address).
- A '.env' file in the working directory is loaded during startup, or the one
- named by IGGY_ENV_PATH.
+ A '.env' file in the working directory or its parents is loaded at startup,
+ or the one named by IGGY_ENV_PATH.
Common examples:
IGGY_PATH=/data/iggy # Data directory
@@ -110,9 +110,9 @@ pub struct Args {
/// already present in the environment, so the flag is equivalent to
/// exporting both by hand and the environment always takes precedence.
///
- /// Only the first creation of the root user reads these values. On an
- /// existing data directory the stored root user is recovered as it is and
- /// the flag has no effect.
+ /// These values initialize only a newly created root user. On restart,
+ /// supplied credentials are validated but do not replace the recovered
+ /// root user or its stored password.
///
/// Examples:
/// iggy-server --with-default-root-credentials # Root logs in as
iggy/iggy
diff --git a/core/server/src/dispatch/partition.rs
b/core/server/src/dispatch/partition.rs
index eadc9d829..11255dcb0 100644
--- a/core/server/src/dispatch/partition.rs
+++ b/core/server/src/dispatch/partition.rs
@@ -255,7 +255,7 @@ where
/// The partition plane admits writes on the primary only (it asserts so), and
a
/// poll is served on whichever node owns the namespace locally, which may be a
/// backup. So gate on primary status here. Auto-commit
-/// is server-managed best-effort (at-least-once delivery), so a
follower-served
+/// is server-managed best-effort, so a follower-served
/// poll simply does not advance the durable offset. The same contract covers a
/// local cursor that never became durable: when the per-kind live map is over
/// its limit the partition evicts such a cursor, and that consumer's next
diff --git a/core/server/src/http/forward.rs b/core/server/src/http/forward.rs
index caf2aa8fa..423f073a4 100644
--- a/core/server/src/http/forward.rs
+++ b/core/server/src/http/forward.rs
@@ -25,12 +25,12 @@
//! them against the current primary's HTTP listener and relays the primary's
//! response on the original connection, so any node answers any request.
//!
-//! Scope: the middleware is attached (via `route_layer`) only to the
-//! control-plane routes, whose ops all commit through the metadata consensus
-//! group and therefore share one forward target. Partition-plane writes
-//! (produce, consumer-offset writes) are excluded: each partition is its own
-//! consensus group whose primary can diverge from the metadata primary, so
-//! forwarding them needs per-group target resolution.
+//! Control-plane routes share the metadata primary as their forward target.
+//! Partition-write routes use a separate fallback: after a typed
+//! `TransientNotAccepted` response, try each other roster node at most once.
+//! That denial proves the operation never entered a partition pipeline.
+//! Partition primaries can differ from the metadata primary, so this fallback
+//! cannot use the metadata leader as its sole target.
//!
//! Safety model, in order:
//! - The bearer is verified locally (verify-only, no session mint) before any
@@ -82,6 +82,7 @@ use crate::http::error::{
CustomError, error_response, gateway_timeout_response,
primary_http_socket, with_retry_after,
};
use crate::http::extractor::{bearer_token, resolve_credential};
+use crate::http::handlers::DURABILITY_HEADER;
use crate::http::state::{APPLIED_OP_HEADER, ForwardState, HttpInner,
VIEW_HEADER};
use crate::server_error::ServerError;
@@ -135,8 +136,15 @@ const RESPONSE_CAPACITY_HINT: usize = 64 * 1024;
/// applied op, not this follower's (the response layer only fills either when
/// absent); the applied op is also what this node records as the caller's
/// read-your-writes floor, so dropping it here would reopen the stale read.
-const RELAYED_RESPONSE_HEADERS: [HeaderName; 4] =
- [CONTENT_TYPE, RETRY_AFTER, VIEW_HEADER, APPLIED_OP_HEADER];
+/// `iggy-durability` preserves the primary's acknowledged completion policy
+/// for writes.
+const RELAYED_RESPONSE_HEADERS: [HeaderName; 5] = [
+ CONTENT_TYPE,
+ RETRY_AFTER,
+ VIEW_HEADER,
+ APPLIED_OP_HEADER,
+ DURABILITY_HEADER,
+];
/// Build the [`ForwardState`] at listener startup.
///
diff --git a/core/server/src/http/handlers.rs b/core/server/src/http/handlers.rs
index be50736a7..cbb505a7f 100644
--- a/core/server/src/http/handlers.rs
+++ b/core/server/src/http/handlers.rs
@@ -168,7 +168,7 @@ const HTTP_READ_CLIENT_ID: u128 = 0;
/// The completed topic policy after an awaited quorum commit. If namespace
/// replacement prevents attesting its incarnation, report the proven quorum
/// guarantee. [`DURABILITY_NONE`] means `?ack=none` dispatch acceptance.
-const DURABILITY_HEADER: HeaderName =
HeaderName::from_static("iggy-durability");
+pub(super) const DURABILITY_HEADER: HeaderName =
HeaderName::from_static("iggy-durability");
const DURABILITY_NONE: &str = "none";
diff --git a/core/server/src/responses.rs b/core/server/src/responses.rs
index c5fca28a9..d963b0f38 100644
--- a/core/server/src/responses.rs
+++ b/core/server/src/responses.rs
@@ -1026,7 +1026,7 @@ fn topic_option_descriptors() ->
Result<Vec<OptionDescriptor>, IggyError> {
.map_err(|_| IggyError::InvalidFormat)?,
kind: HeaderKind::String.as_code(),
default_value: Bytes::from_static(b"none"),
- description: "Compression algorithm (none, gzip)".to_string(),
+ description: "Compression algorithm (none, gzip); stored and
reported, not applied to messages".to_string(),
},
OptionDescriptor {
key: WireName::new(topic_option_keys::MESSAGE_EXPIRY)
@@ -1046,8 +1046,8 @@ fn topic_option_descriptors() ->
Result<Vec<OptionDescriptor>, IggyError> {
default_value: Bytes::copy_from_slice(
&iggy_common::DEFAULT_MAX_TOPIC_SIZE.to_le_bytes(),
),
- description: "Topic size cap in bytes, or a byte-size string (e.g.
1 GiB); \
- must be at least the segment size"
+ description: "Topic-wide sealed-segment size cap, split across all
partitions, in bytes \
+ or a byte-size string (e.g. 1 GiB); finite
values must be at least the segment size"
.to_string(),
},
OptionDescriptor {
@@ -1082,9 +1082,8 @@ fn topic_option_descriptors() ->
Result<Vec<OptionDescriptor>, IggyError> {
&iggy_common::DEFAULT_MESSAGES_REQUIRED_TO_SAVE.to_le_bytes(),
),
description: format!(
- "Flush the journal once it holds this many messages; \
- 1..={}. A threshold no segment can reach leaves committed
\
- messages in the journal, which a crash does not preserve",
+ "Ordinary message-count flush trigger; 1..={}. Required
persistence, \
+ capacity pressure, and lifecycle operations can flush
earlier",
iggy_common::MAX_MESSAGES_REQUIRED_TO_SAVE
),
},
@@ -1109,11 +1108,10 @@ fn topic_option_descriptors() ->
Result<Vec<OptionDescriptor>, IggyError> {
iggy_common::DEFAULT_PREALLOCATE_SEGMENTS,
)]),
description: format!(
- "Reserve each segment's bytes up front where the filesystem
supports \
- it; pairs with segment_size. The reservation is real disk
and runs \
- inline on the owning shard, at every rotation and once
per owned \
- partition at boot, so segment_size * partitions_count is
capped at \
- {} bytes",
+ "Request segment_size bytes of disk reservation when each
segment opens. \
+ Unsupported or failed reservations fall back to ordinary
allocation \
+ with a warning. Reservation runs inline on the owning
shard. \
+ segment_size * partitions_count is capped at {} bytes per
create",
iggy_common::MAX_PREALLOCATED_TOPIC_BYTES
),
},
@@ -1518,7 +1516,7 @@ const SERVER_VERSION: &str = env!("CARGO_PKG_VERSION");
///
/// The SDK strips a result section off exactly the replies whose operation is
/// [`iggy_binary_protocol::Operation::is_result_framed`] (every metadata op
plus the
-/// four consumer-offset ops), and a non-empty `Register`, which it handles on
its
+/// two consumer-offset writes), and a non-empty `Register`, which it handles
on its
/// own. For those, a payload missing the leading zero count has its first
four bytes
/// eaten as a result count, and the decode fails or, worse, succeeds on the
shifted
/// remainder: the raw-PAT reply shipped once without the prefix and broke the
SDK.
diff --git a/core/server/src/shard_allocator.rs
b/core/server/src/shard_allocator.rs
index 98befb1e6..606e64881 100644
--- a/core/server/src/shard_allocator.rs
+++ b/core/server/src/shard_allocator.rs
@@ -21,8 +21,8 @@
//! core so they do not fight over CPU time. This module reads the
//! operator's choice ([`CpuAllocation`] from the config), looks at the
//! real machine with `hwloc`, and hands back one [`ShardInfo`] per
-//! shard. On Linux it also pins each shard's thread to its core and
-//! pins memory to the right NUMA node, so memory stays close and fast.
+//! shard. With pinning enabled on Linux, it binds shard threads to CPUs.
+//! NUMA allocation modes also bind memory to the selected node.
use cpu_allocation::{CpuAllocation, NumaConfig, allowed_cpus};
use hwlocality::Topology;
diff --git a/core/server_common/src/consensus_message.rs
b/core/server_common/src/consensus_message.rs
index c6e428b77..0e4af1e11 100644
--- a/core/server_common/src/consensus_message.rs
+++ b/core/server_common/src/consensus_message.rs
@@ -931,10 +931,8 @@ where
Command::RequestPrepares => Ok(Self::RequestPrepares(
value.try_into_typed::<RequestPreparesHeader>()?,
)),
- // A repaired prepare is a stored PrepareHeader frame whose command
- // byte was rewritten; typed validation would reject the byte, so
- // parse through the generic backing and trust the prepare-shaped
- // layout the way the journal that produced it did.
+ // Keep the repair command through routing so it reaches repair
+ // ingest instead of the live-prepare view fence.
Command::RepairPrepare => Ok(Self::RepairPrepare(
value.try_into_typed::<RepairPrepareHeader>()?,
)),
diff --git a/core/server_common/src/log/logger.rs
b/core/server_common/src/log/logger.rs
index 6ce0f919a..b1d203847 100644
--- a/core/server_common/src/log/logger.rs
+++ b/core/server_common/src/log/logger.rs
@@ -437,17 +437,19 @@ impl Logging {
global::set_tracer_provider(tracer_provider.clone());
global::set_text_map_propagator(TraceContextPropagator::new());
- // Reload telemetry layers with actual implementations
+ // Layer constructors may log, so keep them outside the reload lock.
+ let otel_logs_layer =
OpenTelemetryTracingBridge::new(&logger_provider).boxed();
+ let otel_traces_layer = OpenTelemetryLayer::new(tracer).boxed();
self.otel_logs_reload_handle
.as_ref()
.ok_or(LogError::FilterReloadFailure)?
- .modify(|layer| *layer =
OpenTelemetryTracingBridge::new(&logger_provider).boxed())
+ .modify(|layer| *layer = otel_logs_layer)
.expect("Failed to modify telemetry logs layer");
self.otel_traces_reload_handle
.as_ref()
.ok_or(LogError::FilterReloadFailure)?
- .modify(|layer| *layer = OpenTelemetryLayer::new(tracer).boxed())
+ .modify(|layer| *layer = otel_traces_layer)
.expect("Failed to modify telemetry traces layer");
info!(
diff --git a/core/server_common/src/memory_pool.rs
b/core/server_common/src/memory_pool.rs
index a30ed4b2e..cb3a5f466 100644
--- a/core/server_common/src/memory_pool.rs
+++ b/core/server_common/src/memory_pool.rs
@@ -44,8 +44,8 @@ const BUCKET_SIZES: [usize; NUM_BUCKETS] = [
768 * 1024,
1024 * 1024,
1536 * 1024,
- 2 * 1024 * 1024, // Above 2MiB everything should be rounded up to the next
power of 2 to take advantage of hugepages
- 4 * 1024 * 1024, // (environment variables MIMALLOC_ALLOW_LARGE_OS_PAGES=1
and MIMALLOC_LARGE_OS_PAGES=1).
+ 2 * 1024 * 1024, // Larger buckets use 2 MiB multiples to fit large OS
pages.
+ 4 * 1024 * 1024,
6 * 1024 * 1024,
8 * 1024 * 1024,
10 * 1024 * 1024,
diff --git a/core/server_common/src/send_messages.rs
b/core/server_common/src/send_messages.rs
index 436defa54..539fbf40a 100644
--- a/core/server_common/src/send_messages.rs
+++ b/core/server_common/src/send_messages.rs
@@ -564,8 +564,8 @@ pub fn decode_prepare_slice(bytes: &[u8]) ->
Result<BatchRef<'_>, IggyError> {
/// INVARIANT: `bytes` MUST be node-local self-stamped -
/// [`stamp_prepare_for_persistence`] recomputed the batch checksum over the
/// exact blob on the local node - or already integrity-checked at network
-/// ingress. There is no consensus-layer blob validation: the `PrepareHeader`
-/// integrity fields are inert zeros. Replicated and repaired prepares are
+/// ingress. The partition `PrepareHeader` identity covers only the header; its
+/// body checksum is zero. Replicated and repaired message batches are
/// validated via [`decode_prepare_slice`] before the bytes reach any trusted
/// decode. Calling this on unvalidated network bytes would let a corrupted
blob
/// pass undetected. The full-body per-message checksum pass dominates
