You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/08/17 05:58:59 UTC
(iggy) branch master updated: refactor(configs): clean up server config after full field audit (#3895)
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 338cd2eaf refactor(configs): clean up server config after full field audit (#3895)
338cd2eaf is described below
commit 338cd2eaf7e14e2dc36e102eb6189efad43afb8d
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Aug 17 07:58:54 2026 +0200
refactor(configs): clean up server config after full field audit (#3895)
A field-by-field audit found dead fields with live plumbing,
inert features, unsafe defaults, and stale comments.
Drop dead fields and their plumbing, reject unsignable JWT
algorithms at boot, fix QUIC initial_mtu, split the per-stream
receive window, and enable the message cleaner by default.
Client-listener TLS keeps the on-disk-certificate precedence:
self_signed mints an ephemeral certificate only when cert_file
is absent, and the server logs which way it went. Size
retention divides max_topic_size across partitions, ignores
the active segment, floors each share at one sealed segment
(warned and logged), and re-stages spent cleaner passes so
backlogs drain continuously.
Also bump every SDK touched by the previous release train one
edge higher, and require authentication for the HTTP metrics
endpoint.
Upgrade notes:
- Topics with finite message_expiry or max_topic_size start
deleting oldest sealed segments shortly after upgrade.
- GET /metrics now requires a bearer credential (JWT or PAT);
configure scrapers accordingly (helm serviceMonitor gained an
authorization passthrough).
- QUIC cert path defaults now empty.
- http.jwt.algorithm refuses non-HMAC at boot when HTTP
enabled; trusted issuers unaffected.
---
Cargo.lock | 61 +++--
Cargo.toml | 10 +-
bdd/python/uv.lock | 2 +-
core/ai/mcp/Cargo.toml | 2 +-
core/bench/Cargo.toml | 2 +-
core/bench/dashboard/frontend/Cargo.toml | 2 +-
core/bench/dashboard/server/Cargo.toml | 2 +-
core/bench/report/Cargo.toml | 2 +-
.../bench/src/actors/consumer/client/high_level.rs | 21 +-
core/bench/src/actors/consumer/client/low_level.rs | 11 +-
.../bench/src/actors/producer/client/high_level.rs | 18 +-
core/bench/src/actors/producer/client/low_level.rs | 11 +-
core/bench/src/benchmarks/benchmark.rs | 21 +-
core/bench/src/benchmarks/common.rs | 11 +-
core/bench/src/runner.rs | 9 +-
core/bench/src/utils/client_factory.rs | 16 +-
core/bench/src/utils/mod.rs | 2 +-
core/binary_protocol/Cargo.toml | 2 +-
core/cli/Cargo.toml | 2 +-
core/common/Cargo.toml | 2 +-
core/common/src/utils/topic_size.rs | 7 +-
core/configs/Cargo.toml | 1 -
core/configs/src/common/cache_indexes.rs | 89 -------
core/configs/src/common/defaults.rs | 84 +-----
core/configs/src/common/displays.rs | 50 +---
core/configs/src/common/http.rs | 48 +++-
core/configs/src/common/mod.rs | 1 -
core/configs/src/common/server.rs | 13 -
core/configs/src/common/system.rs | 48 +---
core/configs/src/common/validators.rs | 33 +--
core/configs/src/lib.rs | 2 +-
core/configs/src/server_config/defaults.rs | 49 +---
core/configs/src/server_config/displays.rs | 40 +--
core/configs/src/server_config/partition.rs | 32 ++-
core/configs/src/server_config/quic.rs | 110 +++++---
core/configs/src/server_config/server.rs | 6 +-
core/configs/src/server_config/tcp.rs | 21 --
core/configs/src/server_config/validators.rs | 286 +++++++++++----------
core/connectors/runtime/Cargo.toml | 2 +-
core/connectors/sdk/Cargo.toml | 2 +-
core/connectors/sinks/clickhouse_sink/Cargo.toml | 2 +-
core/connectors/sinks/delta_sink/Cargo.toml | 2 +-
core/connectors/sinks/doris_sink/Cargo.toml | 2 +-
.../connectors/sinks/elasticsearch_sink/Cargo.toml | 2 +-
core/connectors/sinks/http_sink/Cargo.toml | 2 +-
core/connectors/sinks/iceberg_sink/Cargo.toml | 2 +-
core/connectors/sinks/influxdb_sink/Cargo.toml | 2 +-
core/connectors/sinks/meilisearch_sink/Cargo.toml | 2 +-
core/connectors/sinks/mongodb_sink/Cargo.toml | 2 +-
core/connectors/sinks/postgres_sink/Cargo.toml | 2 +-
core/connectors/sinks/quickwit_sink/Cargo.toml | 2 +-
core/connectors/sinks/s3_sink/Cargo.toml | 2 +-
core/connectors/sinks/stdout_sink/Cargo.toml | 2 +-
core/connectors/sinks/surrealdb_sink/Cargo.toml | 2 +-
.../sources/elasticsearch_source/Cargo.toml | 2 +-
core/connectors/sources/influxdb_source/Cargo.toml | 2 +-
core/connectors/sources/postgres_source/Cargo.toml | 2 +-
core/connectors/sources/random_source/Cargo.toml | 2 +-
core/harness_derive/src/attrs.rs | 39 ++-
core/harness_derive/src/codegen.rs | 18 +-
core/harness_derive/src/lib.rs | 10 +-
core/integration/src/harness/config/resolve.rs | 25 +-
core/integration/src/harness/handle/server.rs | 8 +-
core/integration/tests/config_provider/mod.rs | 56 +---
.../data_integrity/verify_after_server_restart.rs | 31 +--
.../verify_consumer_group_partition_assignment.rs | 70 +----
core/integration/tests/sdk/consumer_group.rs | 10 +-
.../tests/sdk/consumer_group_membership.rs | 17 +-
core/integration/tests/server/cg.rs | 21 +-
core/integration/tests/server/flush_vsr.rs | 3 +-
core/integration/tests/server/general.rs | 18 --
core/integration/tests/server/http_client.rs | 3 +-
core/integration/tests/server/http_rbac.rs | 22 +-
core/integration/tests/server/message_cleanup.rs | 12 +
core/integration/tests/server/message_retrieval.rs | 50 +---
.../integration/tests/server/poll_semantics_vsr.rs | 15 +-
core/integration/tests/server/purge_delete.rs | 32 +--
core/integration/tests/server/purge_vsr.rs | 3 +-
.../concurrent_produce_consume_scenario.rs | 2 +-
.../scenarios/cross_protocol_pat_scenario.rs | 15 +-
.../server/scenarios/message_cleanup_scenario.rs | 72 ++++--
.../scenarios/read_during_persistence_scenario.rs | 35 +--
.../scenarios/restart_offset_skip_scenario.rs | 3 +-
.../scenarios/segment_rotation_race_scenario.rs | 1 -
.../stale_client_consumer_group_scenario.rs | 4 -
core/integration/tests/server/specific.rs | 19 +-
core/integration/tests/server/stats_vsr.rs | 3 +-
.../tests/server/topic_admission_vsr.rs | 177 +++++++------
core/message_bus/src/config.rs | 43 ++--
core/message_bus/src/transports/quic.rs | 15 +-
core/message_bus/src/transports/tls/mod.rs | 33 ++-
core/metadata/src/stm/stream.rs | 45 +++-
core/partitions/src/iggy_partition.rs | 276 ++++++++++++++++++--
core/partitions/src/lib.rs | 2 +-
core/sdk/Cargo.toml | 2 +-
core/sdk/src/http/http_client.rs | 1 -
core/server/Cargo.toml | 2 +-
core/server/config.toml | 270 ++++++++-----------
core/server/src/bootstrap.rs | 48 +++-
core/server/src/dispatch.rs | 141 ++++++++--
core/server/src/http.rs | 25 +-
core/server/src/http/handlers.rs | 38 ++-
core/server/src/http/jwt.rs | 22 +-
core/server/src/http/metrics.rs | 12 +-
core/server/src/partition_reconciler.rs | 7 +-
core/server/src/segment_cleaner.rs | 178 +++++++++++--
core/server/src/session_manager.rs | 11 +-
core/server_common/src/messages_batch_mut.rs | 133 ----------
core/shard/src/lib.rs | 28 +-
core/shard/src/router.rs | 31 ++-
examples/python/uv.lock | 2 +-
foreign/csharp/Iggy_SDK/Iggy_SDK.csproj | 2 +-
foreign/go/contracts/version.go | 2 +-
foreign/node/README.md | 8 +-
foreign/node/package-lock.json | 4 +-
foreign/node/package.json | 2 +-
foreign/node/src/client/client.config.test.ts | 35 +++
foreign/node/src/client/client.config.ts | 23 ++
foreign/node/src/client/client.socket.ts | 3 +
foreign/node/src/client/client.type.ts | 7 +-
foreign/node/src/debug.ts | 4 +-
foreign/python/Cargo.toml | 2 +-
foreign/python/pyproject.toml | 2 +-
foreign/python/uv.lock | 2 +-
helm/charts/iggy/README.md | 3 +-
helm/charts/iggy/templates/servicemonitor.yaml | 4 +
helm/charts/iggy/values.yaml | 7 +
web/package-lock.json | 4 +-
web/package.json | 2 +-
129 files changed, 1765 insertions(+), 1679 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 53a0f0e82..7f9abf32d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1950,7 +1950,7 @@ checksum = "3a8241f3ebb85c056b509d4327ad0358fbbba6ffb340bf388f26350aeda225b1"
[[package]]
name = "bench-dashboard-frontend"
-version = "0.8.0-edge.1"
+version = "0.8.0-edge.2"
dependencies = [
"bench-dashboard-shared",
"bench-report",
@@ -1980,7 +1980,7 @@ dependencies = [
[[package]]
name = "bench-report"
-version = "0.4.0-edge.1"
+version = "0.4.0-edge.2"
dependencies = [
"charming",
"colored",
@@ -3196,7 +3196,6 @@ dependencies = [
"serde_with",
"server_common",
"static-toml",
- "strum 0.28.0",
"tracing",
]
@@ -6623,7 +6622,7 @@ checksum = "cd62e6b5e86ea8eeeb8db1de02880a6abc01a397b2ebb64b5d74ac255318f5cb"
[[package]]
name = "iggy"
-version = "0.11.0-edge.2"
+version = "0.11.0-edge.3"
dependencies = [
"async-broadcast",
"async-dropper",
@@ -6657,7 +6656,7 @@ dependencies = [
[[package]]
name = "iggy-bench"
-version = "0.6.0-edge.2"
+version = "0.6.0-edge.3"
dependencies = [
"async-trait",
"bench-report",
@@ -6686,7 +6685,7 @@ dependencies = [
[[package]]
name = "iggy-bench-dashboard-server"
-version = "0.8.0-edge.1"
+version = "0.8.0-edge.2"
dependencies = [
"actix-cors",
"actix-files",
@@ -6714,7 +6713,7 @@ dependencies = [
[[package]]
name = "iggy-cli"
-version = "0.14.0-edge.2"
+version = "0.14.0-edge.3"
dependencies = [
"anyhow",
"apple-native-keyring-store",
@@ -6748,7 +6747,7 @@ dependencies = [
[[package]]
name = "iggy-connectors"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
dependencies = [
"async-trait",
"axum",
@@ -6800,7 +6799,7 @@ dependencies = [
[[package]]
name = "iggy-mcp"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
dependencies = [
"axum",
"axum-server",
@@ -6834,7 +6833,7 @@ dependencies = [
[[package]]
name = "iggy_binary_protocol"
-version = "0.11.0-edge.2"
+version = "0.11.0-edge.3"
dependencies = [
"aligned-vec",
"bytemuck",
@@ -6847,7 +6846,7 @@ dependencies = [
[[package]]
name = "iggy_common"
-version = "0.11.0-edge.2"
+version = "0.11.0-edge.3"
dependencies = [
"aes-gcm",
"async-broadcast",
@@ -6883,7 +6882,7 @@ dependencies = [
[[package]]
name = "iggy_connector_clickhouse_sink"
-version = "0.2.0-edge.2"
+version = "0.2.0-edge.3"
dependencies = [
"async-trait",
"bytes",
@@ -6901,7 +6900,7 @@ dependencies = [
[[package]]
name = "iggy_connector_delta_sink"
-version = "0.2.0-edge.2"
+version = "0.2.0-edge.3"
dependencies = [
"async-trait",
"chrono",
@@ -6918,7 +6917,7 @@ dependencies = [
[[package]]
name = "iggy_connector_doris_sink"
-version = "0.2.0-edge.2"
+version = "0.2.0-edge.3"
dependencies = [
"async-trait",
"base64",
@@ -6938,7 +6937,7 @@ dependencies = [
[[package]]
name = "iggy_connector_elasticsearch_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
dependencies = [
"async-trait",
"base64",
@@ -6956,7 +6955,7 @@ dependencies = [
[[package]]
name = "iggy_connector_elasticsearch_source"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
dependencies = [
"async-trait",
"dashmap",
@@ -6974,7 +6973,7 @@ dependencies = [
[[package]]
name = "iggy_connector_http_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
dependencies = [
"async-trait",
"base64",
@@ -6996,7 +6995,7 @@ dependencies = [
[[package]]
name = "iggy_connector_iceberg_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
dependencies = [
"arrow-json 57.3.1",
"async-trait",
@@ -7015,7 +7014,7 @@ dependencies = [
[[package]]
name = "iggy_connector_influxdb_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
dependencies = [
"async-trait",
"axum",
@@ -7036,7 +7035,7 @@ dependencies = [
[[package]]
name = "iggy_connector_influxdb_source"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
dependencies = [
"ahash 0.8.12",
"async-trait",
@@ -7062,7 +7061,7 @@ dependencies = [
[[package]]
name = "iggy_connector_meilisearch_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
dependencies = [
"async-trait",
"base64",
@@ -7080,7 +7079,7 @@ dependencies = [
[[package]]
name = "iggy_connector_mongodb_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
dependencies = [
"async-trait",
"humantime",
@@ -7096,7 +7095,7 @@ dependencies = [
[[package]]
name = "iggy_connector_postgres_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
dependencies = [
"async-trait",
"dashmap",
@@ -7115,7 +7114,7 @@ dependencies = [
[[package]]
name = "iggy_connector_postgres_source"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
dependencies = [
"async-trait",
"base64",
@@ -7137,7 +7136,7 @@ dependencies = [
[[package]]
name = "iggy_connector_quickwit_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
dependencies = [
"async-trait",
"dashmap",
@@ -7151,7 +7150,7 @@ dependencies = [
[[package]]
name = "iggy_connector_random_source"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
dependencies = [
"async-trait",
"dashmap",
@@ -7168,7 +7167,7 @@ dependencies = [
[[package]]
name = "iggy_connector_s3_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
dependencies = [
"async-trait",
"base64",
@@ -7189,7 +7188,7 @@ dependencies = [
[[package]]
name = "iggy_connector_sdk"
-version = "0.4.0-edge.1"
+version = "0.4.0-edge.2"
dependencies = [
"anyhow",
"apache-avro",
@@ -7225,7 +7224,7 @@ dependencies = [
[[package]]
name = "iggy_connector_stdout_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
dependencies = [
"async-trait",
"dashmap",
@@ -7237,7 +7236,7 @@ dependencies = [
[[package]]
name = "iggy_connector_surrealdb_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
dependencies = [
"async-trait",
"base64",
@@ -11840,7 +11839,7 @@ dependencies = [
[[package]]
name = "server"
-version = "0.9.0-edge.3"
+version = "0.9.0-edge.4"
dependencies = [
"ahash 0.8.12",
"argon2",
diff --git a/Cargo.toml b/Cargo.toml
index 18a95963f..7adc072b3 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -199,11 +199,11 @@ hyper-util = { version = "0.1.20", features = ["server-auto", "service"] }
iceberg = "0.9.1"
iceberg-catalog-rest = "0.9.1"
iceberg-storage-opendal = "0.9.1"
-iggy = { path = "core/sdk", version = "0.11.0-edge.2" }
-iggy-cli = { path = "core/cli", version = "0.14.0-edge.2" }
-iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.0-edge.2" }
-iggy_common = { path = "core/common", version = "0.11.0-edge.2" }
-iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.4.0-edge.1" }
+iggy = { path = "core/sdk", version = "0.11.0-edge.3" }
+iggy-cli = { path = "core/cli", version = "0.14.0-edge.3" }
+iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.0-edge.3" }
+iggy_common = { path = "core/common", version = "0.11.0-edge.3" }
+iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.4.0-edge.2" }
indexmap = "2.14.0"
integration = { path = "core/integration" }
ipnet = "2.12.0"
diff --git a/bdd/python/uv.lock b/bdd/python/uv.lock
index d050315e1..a5ebae63e 100644
--- a/bdd/python/uv.lock
+++ b/bdd/python/uv.lock
@@ -8,7 +8,7 @@ exclude-newer-span = "P7D"
[[package]]
name = "apache-iggy"
-version = "0.9.0.dev2"
+version = "0.9.0.dev3"
source = { directory = "../../foreign/python" }
[package.metadata]
diff --git a/core/ai/mcp/Cargo.toml b/core/ai/mcp/Cargo.toml
index e58068e14..0d061d173 100644
--- a/core/ai/mcp/Cargo.toml
+++ b/core/ai/mcp/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-mcp"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
description = "MCP Server for Iggy message streaming platform"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/bench/Cargo.toml b/core/bench/Cargo.toml
index 1ac1b9baa..d9e3a57da 100644
--- a/core/bench/Cargo.toml
+++ b/core/bench/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-bench"
-version = "0.6.0-edge.2"
+version = "0.6.0-edge.3"
edition = "2024"
license = "Apache-2.0"
repository = "https://github.com/apache/iggy"
diff --git a/core/bench/dashboard/frontend/Cargo.toml b/core/bench/dashboard/frontend/Cargo.toml
index 7c9be3579..0cc89695e 100644
--- a/core/bench/dashboard/frontend/Cargo.toml
+++ b/core/bench/dashboard/frontend/Cargo.toml
@@ -18,7 +18,7 @@
[package]
name = "bench-dashboard-frontend"
license = "Apache-2.0"
-version = "0.8.0-edge.1"
+version = "0.8.0-edge.2"
edition = "2024"
publish = false
diff --git a/core/bench/dashboard/server/Cargo.toml b/core/bench/dashboard/server/Cargo.toml
index dc61bf618..a57782a89 100644
--- a/core/bench/dashboard/server/Cargo.toml
+++ b/core/bench/dashboard/server/Cargo.toml
@@ -18,7 +18,7 @@
[package]
name = "iggy-bench-dashboard-server"
license = "Apache-2.0"
-version = "0.8.0-edge.1"
+version = "0.8.0-edge.2"
edition = "2024"
publish = false
diff --git a/core/bench/report/Cargo.toml b/core/bench/report/Cargo.toml
index a4eb23e89..5aa106554 100644
--- a/core/bench/report/Cargo.toml
+++ b/core/bench/report/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "bench-report"
-version = "0.4.0-edge.1"
+version = "0.4.0-edge.2"
edition = "2024"
description = "Benchmark report and chart generation library for iggy-bench binary and iggy-benchmarks-dashboard web app"
license = "Apache-2.0"
diff --git a/core/bench/src/actors/consumer/client/high_level.rs b/core/bench/src/actors/consumer/client/high_level.rs
index 7277c4046..00128ca26 100644
--- a/core/bench/src/actors/consumer/client/high_level.rs
+++ b/core/bench/src/actors/consumer/client/high_level.rs
@@ -23,7 +23,7 @@ use crate::actors::{
},
};
-use crate::utils::{ClientFactory, authenticate};
+use crate::utils::ClientFactory;
use futures_util::StreamExt;
use iggy::prelude::*;
use std::{sync::Arc, time::Duration};
@@ -33,6 +33,14 @@ use tracing::{error, warn};
pub struct HighLevelConsumerClient {
client_factory: Arc<dyn ClientFactory>,
config: BenchmarkConsumerConfig,
+ /// `IggyConsumer` only borrows the transport, so the owning client has to
+ /// outlive it: dropping the client aborts the SDK heartbeat task, and the
+ /// consumer stops being pinged for the rest of the run. Boxed because the
+ /// low-level client is the smaller of `TypedBenchmarkConsumer`'s two
+ /// variants, and inlining the client here spreads them past what
+ /// `clippy::large_enum_variant` allows.
+ #[allow(dead_code)]
+ client: Option<Box<IggyClient>>,
consumer: Option<IggyConsumer>,
}
@@ -41,6 +49,7 @@ impl HighLevelConsumerClient {
Self {
client_factory,
config,
+ client: None,
consumer: None,
}
}
@@ -101,14 +110,7 @@ impl ConsumerClient for HighLevelConsumerClient {
impl BenchmarkInit for HighLevelConsumerClient {
async fn setup(&mut self) -> Result<(), IggyError> {
let topic_id_str = "topic-1";
- let client = self.client_factory.create_client().await;
- let client = IggyClient::create(client, None, None);
- authenticate(
- &client,
- self.client_factory.username(),
- self.client_factory.password(),
- )
- .await;
+ let client = self.client_factory.create_authenticated_client().await?;
let stream_id_str = self.config.stream_id.clone();
@@ -140,6 +142,7 @@ impl BenchmarkInit for HighLevelConsumerClient {
consumer.init().await?;
self.consumer = Some(consumer);
+ self.client = Some(Box::new(client));
Ok(())
}
}
diff --git a/core/bench/src/actors/consumer/client/low_level.rs b/core/bench/src/actors/consumer/client/low_level.rs
index fc8c1bc64..7f4774895 100644
--- a/core/bench/src/actors/consumer/client/low_level.rs
+++ b/core/bench/src/actors/consumer/client/low_level.rs
@@ -19,7 +19,7 @@ use crate::actors::consumer::client::BenchmarkConsumerClient;
use crate::actors::consumer::client::interface::{BenchmarkConsumerConfig, ConsumerClient};
use crate::actors::{ApiLabel, BatchMetrics, BenchmarkInit};
use crate::benchmarks::common::create_consumer;
-use crate::utils::{ClientFactory, authenticate};
+use crate::utils::ClientFactory;
use crate::utils::{batch_total_size_bytes, batch_user_size_bytes};
use iggy::prelude::*;
use std::sync::Arc;
@@ -118,14 +118,7 @@ impl BenchmarkInit for LowLevelConsumerClient {
let topic_id_str = "topic-1";
let default_partition_id = 0u32;
- let client = self.client_factory.create_client().await;
- let client = IggyClient::create(client, None, None);
- authenticate(
- &client,
- self.client_factory.username(),
- self.client_factory.password(),
- )
- .await;
+ let client = self.client_factory.create_authenticated_client().await?;
let stream_id: Identifier = self.config.stream_id.as_str().try_into().unwrap();
let topic_id = topic_id_str.try_into().unwrap();
diff --git a/core/bench/src/actors/producer/client/high_level.rs b/core/bench/src/actors/producer/client/high_level.rs
index 3db74f869..36ff273ca 100644
--- a/core/bench/src/actors/producer/client/high_level.rs
+++ b/core/bench/src/actors/producer/client/high_level.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use crate::utils::{ClientFactory, authenticate};
+use crate::utils::ClientFactory;
use crate::{
actors::{
ApiLabel, BatchMetrics, BenchmarkInit,
@@ -33,6 +33,11 @@ use tokio::time::Instant;
pub struct HighLevelProducerClient {
client_factory: Arc<dyn ClientFactory>,
config: BenchmarkProducerConfig,
+ /// `IggyProducer` only borrows the transport, so the owning client has to
+ /// outlive it: dropping the client aborts the SDK heartbeat task, and the
+ /// producer stops being pinged for the rest of the run.
+ #[allow(dead_code)]
+ client: Option<IggyClient>,
producer: Option<IggyProducer>,
}
@@ -41,6 +46,7 @@ impl HighLevelProducerClient {
Self {
client_factory,
config,
+ client: None,
producer: None,
}
}
@@ -81,14 +87,7 @@ impl BenchmarkInit for HighLevelProducerClient {
let topic_id_str = "topic-1";
let default_partition_id = 0u32;
- let client = self.client_factory.create_client().await;
- let client = IggyClient::create(client, None, None);
- authenticate(
- &client,
- self.client_factory.username(),
- self.client_factory.password(),
- )
- .await;
+ let client = self.client_factory.create_authenticated_client().await?;
let stream_id_str = self.config.stream_id.clone();
@@ -111,6 +110,7 @@ impl BenchmarkInit for HighLevelProducerClient {
.build(),
);
self.producer.as_mut().unwrap().init().await?;
+ self.client = Some(client);
Ok(())
}
}
diff --git a/core/bench/src/actors/producer/client/low_level.rs b/core/bench/src/actors/producer/client/low_level.rs
index 4831e527a..6e9db0048 100644
--- a/core/bench/src/actors/producer/client/low_level.rs
+++ b/core/bench/src/actors/producer/client/low_level.rs
@@ -17,7 +17,7 @@
use std::sync::Arc;
-use crate::utils::{ClientFactory, authenticate};
+use crate::utils::ClientFactory;
use crate::{
actors::{
ApiLabel, BatchMetrics, BenchmarkInit,
@@ -89,14 +89,7 @@ impl BenchmarkInit for LowLevelProducerClient {
let default_partition_id = 0u32;
let partitions = self.config.partitions;
- let client = self.client_factory.create_client().await;
- let client = IggyClient::create(client, None, None);
- authenticate(
- &client,
- self.client_factory.username(),
- self.client_factory.password(),
- )
- .await;
+ let client = self.client_factory.create_authenticated_client().await?;
let partitioning = match partitions {
0 => panic!("Partition count must be greater than 0"),
diff --git a/core/bench/src/benchmarks/benchmark.rs b/core/bench/src/benchmarks/benchmark.rs
index 3d61bc331..60675237a 100644
--- a/core/bench/src/benchmarks/benchmark.rs
+++ b/core/bench/src/benchmarks/benchmark.rs
@@ -16,12 +16,11 @@
// under the License.
use crate::args::kind::BenchmarkKindCommand;
-use crate::utils::{ClientFactory, authenticate};
+use crate::utils::ClientFactory;
use crate::{args::common::IggyBenchArgs, utils::client_factory::create_client_factory};
use async_trait::async_trait;
use bench_report::benchmark_kind::BenchmarkKind;
use bench_report::individual_metrics::BenchmarkIndividualMetrics;
-use iggy::clients::client::IggyClient;
use iggy::prelude::*;
use std::sync::Arc;
use tokio::task::JoinSet;
@@ -96,14 +95,7 @@ pub trait Benchmarkable: Send {
async fn init_streams(&self) -> Result<(), IggyError> {
let number_of_streams = self.args().streams();
let partitions_count: u32 = self.args().number_of_partitions();
- let client = self.client_factory().create_client().await;
- let client = IggyClient::create(client, None, None);
- authenticate(
- &client,
- self.client_factory().username(),
- self.client_factory().password(),
- )
- .await;
+ let client = self.client_factory().create_authenticated_client().await?;
let reuse_streams = self.args().reuse_streams();
let streams = client.get_streams().await?;
for i in 1..=number_of_streams {
@@ -158,14 +150,7 @@ pub trait Benchmarkable: Send {
async fn check_streams(&self) -> Result<(), IggyError> {
let number_of_streams = self.args().streams();
- let client = self.client_factory().create_client().await;
- let client = IggyClient::create(client, None, None);
- authenticate(
- &client,
- self.client_factory().username(),
- self.client_factory().password(),
- )
- .await;
+ let client = self.client_factory().create_authenticated_client().await?;
let streams = client.get_streams().await?;
for i in 1..=number_of_streams {
let stream_name = format!("bench-stream-{i}");
diff --git a/core/bench/src/benchmarks/common.rs b/core/bench/src/benchmarks/common.rs
index cb5ee9f9a..bd5a38247 100644
--- a/core/bench/src/benchmarks/common.rs
+++ b/core/bench/src/benchmarks/common.rs
@@ -16,7 +16,7 @@
// under the License.
use super::{CONSUMER_GROUP_BASE_ID, CONSUMER_GROUP_NAME_PREFIX};
-use crate::utils::{ClientFactory, authenticate};
+use crate::utils::ClientFactory;
use crate::{
actors::{
consumer::typed_benchmark_consumer::TypedBenchmarkConsumer,
@@ -71,16 +71,9 @@ pub async fn init_consumer_groups(
client_factory: &Arc<dyn ClientFactory>,
args: &IggyBenchArgs,
) -> Result<(), IggyError> {
- let client = client_factory.create_client().await;
- let client = IggyClient::create(client, None, None);
+ let client = client_factory.create_authenticated_client().await?;
let cg_count = args.number_of_consumer_groups();
- authenticate(
- &client,
- client_factory.username(),
- client_factory.password(),
- )
- .await;
for i in 1..=cg_count {
let consumer_group_id = CONSUMER_GROUP_BASE_ID + i;
let stream_name = format!("bench-stream-{i}");
diff --git a/core/bench/src/runner.rs b/core/bench/src/runner.rs
index 49f9e8bfe..f93574efd 100644
--- a/core/bench/src/runner.rs
+++ b/core/bench/src/runner.rs
@@ -22,7 +22,7 @@ use crate::plot::{ChartType, plot_chart};
use crate::utils::cpu_name::append_cpu_name_lowercase;
use crate::utils::{collect_server_logs_and_save_to_file, params_from_args_and_metrics};
use bench_report::hardware::BenchmarkHardware;
-use iggy::prelude::{Client, IggyClient, IggyError, UserClient};
+use iggy::prelude::IggyError;
use std::path::Path;
use std::time::Duration;
use tokio::time::sleep;
@@ -64,12 +64,7 @@ impl BenchmarkRunner {
info!("All actors joined!");
let client_factory = benchmark.client_factory();
- let admin_client_wrapper = client_factory.create_client().await;
- let admin_client = IggyClient::create(admin_client_wrapper, None, None);
- admin_client.connect().await?;
- admin_client
- .login_user(client_factory.username(), client_factory.password())
- .await?;
+ let admin_client = client_factory.create_authenticated_client().await?;
let hardware =
BenchmarkHardware::get_system_info_with_identifier(benchmark.args().identifier());
diff --git a/core/bench/src/utils/client_factory.rs b/core/bench/src/utils/client_factory.rs
index f7f325e89..82d7f57d7 100644
--- a/core/bench/src/utils/client_factory.rs
+++ b/core/bench/src/utils/client_factory.rs
@@ -20,7 +20,7 @@ use crate::args::transport::BenchmarkTransportCommand;
use async_trait::async_trait;
use iggy::http::http_client::HttpClient;
use iggy::prelude::{
- Client, ClientWrapper, HttpClientConfig, IdentityInfo, IggyClient, QuicClientConfig, TcpClient,
+ Client, ClientWrapper, HttpClientConfig, IggyClient, IggyError, QuicClientConfig, TcpClient,
TcpClientConfig, TransportProtocol, UserClient, WebSocketClientConfig,
};
use iggy::quic::quic_client::QuicClient;
@@ -34,10 +34,18 @@ pub trait ClientFactory: Sync + Send {
fn server_addr(&self) -> String;
fn username(&self) -> &str;
fn password(&self) -> &str;
-}
-pub async fn authenticate(client: &IggyClient, username: &str, password: &str) -> IdentityInfo {
- client.login_user(username, password).await.unwrap()
+ /// Connects and logs in, the way an SDK user would.
+ ///
+ /// `IggyClient::connect` spawns the heartbeat task, which also refreshes
+ /// consumer-group assignments. The returned client owns that task, so
+ /// dropping it aborts both for the rest of the run.
+ async fn create_authenticated_client(&self) -> Result<IggyClient, IggyError> {
+ let client = IggyClient::create(self.create_client().await, None, None);
+ client.connect().await?;
+ client.login_user(self.username(), self.password()).await?;
+ Ok(client)
+ }
}
#[derive(Debug, Clone)]
diff --git a/core/bench/src/utils/mod.rs b/core/bench/src/utils/mod.rs
index d4d9f3753..973271f66 100644
--- a/core/bench/src/utils/mod.rs
+++ b/core/bench/src/utils/mod.rs
@@ -36,7 +36,7 @@ use crate::args::{
},
};
-pub use client_factory::{ClientFactory, authenticate};
+pub use client_factory::ClientFactory;
pub mod batch_generator;
pub mod client_factory;
diff --git a/core/binary_protocol/Cargo.toml b/core/binary_protocol/Cargo.toml
index 0d8b74da5..a52840d5a 100644
--- a/core/binary_protocol/Cargo.toml
+++ b/core/binary_protocol/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_binary_protocol"
-version = "0.11.0-edge.2"
+version = "0.11.0-edge.3"
description = "Wire protocol types and codec for the Iggy binary protocol. Shared between server and SDK."
edition = "2024"
rust-version.workspace = true
diff --git a/core/cli/Cargo.toml b/core/cli/Cargo.toml
index 2a6841dd7..580ae5779 100644
--- a/core/cli/Cargo.toml
+++ b/core/cli/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-cli"
-version = "0.14.0-edge.2"
+version = "0.14.0-edge.3"
edition = "2024"
rust-version.workspace = true
authors = ["bartosz.ciesla@gmail.com"]
diff --git a/core/common/Cargo.toml b/core/common/Cargo.toml
index a521aade5..59dd94957 100644
--- a/core/common/Cargo.toml
+++ b/core/common/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_common"
-version = "0.11.0-edge.2"
+version = "0.11.0-edge.3"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
rust-version.workspace = true
diff --git a/core/common/src/utils/topic_size.rs b/core/common/src/utils/topic_size.rs
index 487d208f0..1cfe4c13b 100644
--- a/core/common/src/utils/topic_size.rs
+++ b/core/common/src/utils/topic_size.rs
@@ -29,7 +29,12 @@ pub enum MaxTopicSize {
#[default]
/// Use the default size set by the server
ServerDefault,
- /// Use a custom size
+ /// Best-effort cap on the SEALED bytes a topic retains, enforced per
+ /// partition (the cap divided by the partition count) and raised to one
+ /// sealed segment when set below it, which the server logs. A sealed
+ /// segment can run one maximum message past the topic's `segment_size`, so
+ /// the cap is never a hard disk bound: the active segment is never trimmed
+ /// and a lagging committed consumer pins the sealed ones behind it.
Custom(IggyByteSize),
/// Use an unlimited size
Unlimited,
diff --git a/core/configs/Cargo.toml b/core/configs/Cargo.toml
index 0b2f2079a..d195feed8 100644
--- a/core/configs/Cargo.toml
+++ b/core/configs/Cargo.toml
@@ -35,5 +35,4 @@ serde_json = { workspace = true }
serde_with = { workspace = true }
server_common = { workspace = true }
static-toml = { workspace = true }
-strum = { workspace = true }
tracing = { workspace = true }
diff --git a/core/configs/src/common/cache_indexes.rs b/core/configs/src/common/cache_indexes.rs
deleted file mode 100644
index 0f3939d5e..000000000
--- a/core/configs/src/common/cache_indexes.rs
+++ /dev/null
@@ -1,89 +0,0 @@
-// 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 serde::{Deserialize, Serialize};
-use serde_with::serde_as;
-use std::str::FromStr;
-use strum::Display;
-
-#[serde_as]
-#[derive(Debug, Serialize, Display, Default, PartialEq)]
-#[serde(rename_all = "lowercase")]
-pub enum CacheIndexesConfig {
- All,
- #[default]
- #[serde(rename = "open_segment")]
- OpenSegment,
- None,
-}
-
-impl<'de> Deserialize<'de> for CacheIndexesConfig {
- fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
- where
- D: serde::Deserializer<'de>,
- {
- #[derive(Deserialize)]
- #[serde(untagged)]
- enum StringOrBool {
- String(String),
- Bool(bool),
- }
-
- let value = StringOrBool::deserialize(deserializer)?;
-
- match value {
- StringOrBool::String(s) => match s.to_lowercase().as_str() {
- "all" | "true" => Ok(CacheIndexesConfig::All),
- "open_segment" => Ok(CacheIndexesConfig::OpenSegment),
- "none" | "false" => Ok(CacheIndexesConfig::None),
- _ => Err(serde::de::Error::custom(format!(
- "Invalid CacheIndexesConfig value: {s}"
- ))),
- },
- StringOrBool::Bool(b) => {
- if b {
- Ok(CacheIndexesConfig::All)
- } else {
- Ok(CacheIndexesConfig::None)
- }
- }
- }
- }
-}
-
-impl FromStr for CacheIndexesConfig {
- type Err = String;
-
- fn from_str(s: &str) -> Result<Self, Self::Err> {
- match s.to_lowercase().as_str() {
- "all" | "true" => Ok(CacheIndexesConfig::All),
- "open_segment" => Ok(CacheIndexesConfig::OpenSegment),
- "none" | "false" => Ok(CacheIndexesConfig::None),
- _ => Err(format!("Invalid CacheIndexesConfig: {s}")),
- }
- }
-}
-
-impl From<bool> for CacheIndexesConfig {
- fn from(value: bool) -> Self {
- if value {
- CacheIndexesConfig::All
- } else {
- CacheIndexesConfig::None
- }
- }
-}
diff --git a/core/configs/src/common/defaults.rs b/core/configs/src/common/defaults.rs
index e733c5b7c..1dddfcc54 100644
--- a/core/configs/src/common/defaults.rs
+++ b/core/configs/src/common/defaults.rs
@@ -17,14 +17,13 @@
use super::http::{HttpConfig, HttpCorsConfig, HttpJwtConfig, HttpMetricsConfig, HttpTlsConfig};
use super::server::{
- ConsumerGroupConfig, HeartbeatConfig, MemoryPoolConfig, MessageSaverConfig,
- MessagesMaintenanceConfig, PersonalAccessTokenCleanerConfig, PersonalAccessTokenConfig,
- TelemetryConfig, TelemetryLogsConfig, TelemetryTracesConfig,
+ ConsumerGroupConfig, HeartbeatConfig, MemoryPoolConfig, MessagesMaintenanceConfig,
+ PersonalAccessTokenCleanerConfig, PersonalAccessTokenConfig, TelemetryConfig,
+ TelemetryLogsConfig, TelemetryTracesConfig,
};
use super::system::{
- BackupConfig, CompatibilityConfig, CompressionConfig, EncryptionConfig, LoggingConfig,
- PartitionConfig, RecoveryConfig, RuntimeConfig, SegmentConfig, StateConfig, StreamConfig,
- SystemConfig, TopicConfig,
+ EncryptionConfig, LoggingConfig, PartitionConfig, RecoveryConfig, RuntimeConfig, SegmentConfig,
+ StreamConfig, SystemConfig, TopicConfig,
};
use configs::ConfigEnvMappings;
@@ -150,16 +149,6 @@ impl Default for HttpTlsConfig {
}
}
-impl Default for MessageSaverConfig {
- fn default() -> MessageSaverConfig {
- MessageSaverConfig {
- enabled: SERVER_CONFIG.message_saver.enabled,
- enforce_fsync: SERVER_CONFIG.message_saver.enforce_fsync,
- interval: SERVER_CONFIG.message_saver.interval.parse().unwrap(),
- }
- }
-}
-
impl Default for PersonalAccessTokenConfig {
fn default() -> PersonalAccessTokenConfig {
PersonalAccessTokenConfig {
@@ -187,7 +176,6 @@ impl<S: ConfigEnvMappings + Default> Default for SystemConfig<S> {
fn default() -> Self {
Self {
path: SERVER_CONFIG.system.path.parse().unwrap(),
- backup: BackupConfig::default(),
runtime: RuntimeConfig::default(),
logging: LoggingConfig::default(),
stream: StreamConfig::default(),
@@ -195,8 +183,6 @@ impl<S: ConfigEnvMappings + Default> Default for SystemConfig<S> {
topic: TopicConfig::default(),
partition: PartitionConfig::default(),
segment: SegmentConfig::default(),
- state: StateConfig::default(),
- compression: CompressionConfig::default(),
recovery: RecoveryConfig::default(),
memory_pool: MemoryPoolConfig::default(),
sharding: S::default(),
@@ -204,29 +190,6 @@ impl<S: ConfigEnvMappings + Default> Default for SystemConfig<S> {
}
}
-impl Default for BackupConfig {
- fn default() -> BackupConfig {
- BackupConfig {
- path: SERVER_CONFIG.system.backup.path.parse().unwrap(),
- compatibility: CompatibilityConfig::default(),
- }
- }
-}
-
-impl Default for CompatibilityConfig {
- fn default() -> Self {
- CompatibilityConfig {
- path: SERVER_CONFIG
- .system
- .backup
- .compatibility
- .path
- .parse()
- .unwrap(),
- }
- }
-}
-
impl Default for HeartbeatConfig {
fn default() -> HeartbeatConfig {
HeartbeatConfig {
@@ -244,11 +207,6 @@ impl Default for ConsumerGroupConfig {
.rebalancing_timeout
.parse()
.unwrap(),
- rebalancing_check_interval: SERVER_CONFIG
- .consumer_group
- .rebalancing_check_interval
- .parse()
- .unwrap(),
}
}
}
@@ -261,20 +219,6 @@ impl Default for RuntimeConfig {
}
}
-impl Default for CompressionConfig {
- fn default() -> Self {
- CompressionConfig {
- allow_override: SERVER_CONFIG.system.compression.allow_override,
- default_algorithm: SERVER_CONFIG
- .system
- .compression
- .default_algorithm
- .parse()
- .unwrap(),
- }
- }
-}
-
impl Default for LoggingConfig {
fn default() -> LoggingConfig {
LoggingConfig {
@@ -290,12 +234,6 @@ impl Default for LoggingConfig {
.parse()
.unwrap(),
retention: SERVER_CONFIG.system.logging.retention.parse().unwrap(),
- sysinfo_print_interval: SERVER_CONFIG
- .system
- .logging
- .sysinfo_print_interval
- .parse()
- .unwrap(),
}
}
}
@@ -337,23 +275,11 @@ impl Default for PartitionConfig {
impl Default for SegmentConfig {
fn default() -> SegmentConfig {
SegmentConfig {
- cache_indexes: SERVER_CONFIG.system.segment.cache_indexes.parse().unwrap(),
archive_expired: SERVER_CONFIG.system.segment.archive_expired,
}
}
}
-impl Default for StateConfig {
- fn default() -> StateConfig {
- StateConfig {
- enforce_fsync: SERVER_CONFIG.system.state.enforce_fsync,
- max_file_operation_retries: SERVER_CONFIG.system.state.max_file_operation_retries
- as u32,
- retry_delay: SERVER_CONFIG.system.state.retry_delay.parse().unwrap(),
- }
- }
-}
-
impl Default for RecoveryConfig {
fn default() -> RecoveryConfig {
RecoveryConfig {
diff --git a/core/configs/src/common/displays.rs b/core/configs/src/common/displays.rs
index 234cc0412..32af51798 100644
--- a/core/configs/src/common/displays.rs
+++ b/core/configs/src/common/displays.rs
@@ -21,10 +21,9 @@ use super::server::{
};
use super::{
http::{HttpConfig, HttpCorsConfig, HttpJwtConfig, HttpMetricsConfig, HttpTlsConfig},
- server::MessageSaverConfig,
system::{
- CompressionConfig, EncryptionConfig, LoggingConfig, PartitionConfig, SegmentConfig,
- StateConfig, StreamConfig, SystemConfig, TopicConfig,
+ EncryptionConfig, LoggingConfig, PartitionConfig, SegmentConfig, StreamConfig,
+ SystemConfig, TopicConfig,
},
};
use configs::ConfigEnvMappings;
@@ -93,16 +92,6 @@ impl Display for HttpTlsConfig {
}
}
-impl Display for CompressionConfig {
- fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
- write!(
- f,
- "{{ allowed_override: {}, default_algorithm: {} }}",
- self.allow_override, self.default_algorithm
- )
- }
-}
-
impl Display for DataMaintenanceConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(f, "{{ messages: {} }}", self.messages)
@@ -121,21 +110,7 @@ impl Display for MessagesMaintenanceConfig {
impl Display for ConsumerGroupConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
- write!(
- f,
- "{{ rebalancing_timeout: {}, rebalancing_check_interval: {} }}",
- self.rebalancing_timeout, self.rebalancing_check_interval
- )
- }
-}
-
-impl Display for MessageSaverConfig {
- fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
- write!(
- f,
- "{{ enabled: {}, enforce_fsync: {}, interval: {} }}",
- self.enabled, self.enforce_fsync, self.interval
- )
+ write!(f, "{{ rebalancing_timeout: {} }}", self.rebalancing_timeout)
}
}
@@ -179,11 +154,7 @@ impl Display for PartitionConfig {
impl Display for SegmentConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
- write!(
- f,
- "{{ cache_indexes: {}, archive_expired: {} }}",
- self.cache_indexes, self.archive_expired,
- )
+ write!(f, "{{ archive_expired: {} }}", self.archive_expired,)
}
}
@@ -223,16 +194,6 @@ impl Display for TelemetryLogsConfig {
}
}
-impl Display for StateConfig {
- fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
- write!(
- f,
- "{{ enforce_fsync: {}, max_file_operation_retries: {}, retry_delay: {} }}",
- self.enforce_fsync, self.max_file_operation_retries, self.retry_delay,
- )
- }
-}
-
impl Display for TelemetryTracesConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
@@ -247,7 +208,7 @@ impl<S: ConfigEnvMappings> Display for SystemConfig<S> {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
- "{{ path: {}, logging: {}, stream: {}, topic: {}, partition: {}, segment: {}, encryption: {}, state: {} }}",
+ "{{ path: {}, logging: {}, stream: {}, topic: {}, partition: {}, segment: {}, encryption: {} }}",
self.path,
self.logging,
self.stream,
@@ -255,7 +216,6 @@ impl<S: ConfigEnvMappings> Display for SystemConfig<S> {
self.partition,
self.segment,
self.encryption,
- self.state,
)
}
}
diff --git a/core/configs/src/common/http.rs b/core/configs/src/common/http.rs
index cb6271d32..625478097 100644
--- a/core/configs/src/common/http.rs
+++ b/core/configs/src/common/http.rs
@@ -25,6 +25,19 @@ use serde::{Deserialize, Serialize};
use serde_with::DisplayFromStr;
use serde_with::serde_as;
+/// Every `http.jwt.algorithm` the server can sign and verify with, and the
+/// `jsonwebtoken` algorithm each names.
+///
+/// HMAC only, because [`HttpJwtConfig::get_encoding_key`] builds its key from a
+/// shared secret and there is no PEM loading path. `jsonwebtoken::encode`
+/// refuses a header algorithm whose family does not match the key, so an
+/// asymmetric algorithm would boot clean and then fail every login.
+pub const HMAC_JWT_ALGORITHMS: [(&str, Algorithm); 3] = [
+ ("HS256", Algorithm::HS256),
+ ("HS384", Algorithm::HS384),
+ ("HS512", Algorithm::HS512),
+];
+
#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct TrustedIssuerConfig {
pub issuer: String,
@@ -110,15 +123,11 @@ pub struct HttpTlsConfig {
impl HttpJwtConfig {
pub fn get_algorithm(&self) -> Result<Algorithm, IggyError> {
- match self.algorithm.as_str() {
- "HS256" => Ok(Algorithm::HS256),
- "HS384" => Ok(Algorithm::HS384),
- "HS512" => Ok(Algorithm::HS512),
- "RS256" => Ok(Algorithm::RS256),
- "RS384" => Ok(Algorithm::RS384),
- "RS512" => Ok(Algorithm::RS512),
- _ => Err(IggyError::InvalidJwtAlgorithm(self.algorithm.clone())),
- }
+ HMAC_JWT_ALGORITHMS
+ .iter()
+ .find(|(name, _)| *name == self.algorithm)
+ .map(|(_, algorithm)| *algorithm)
+ .ok_or_else(|| IggyError::InvalidJwtAlgorithm(self.algorithm.clone()))
}
pub fn get_decoding_secret(&self) -> JwtSecret {
@@ -168,6 +177,27 @@ impl HttpJwtConfig {
mod tests {
use super::*;
+ #[test]
+ fn get_algorithm_accepts_only_hmac() {
+ for (name, expected) in HMAC_JWT_ALGORITHMS {
+ let config = HttpJwtConfig {
+ algorithm: name.to_string(),
+ ..HttpJwtConfig::default()
+ };
+ assert_eq!(config.get_algorithm(), Ok(expected));
+ }
+ for asymmetric in ["RS256", "RS384", "RS512", "ES256", "EdDSA"] {
+ let config = HttpJwtConfig {
+ algorithm: asymmetric.to_string(),
+ ..HttpJwtConfig::default()
+ };
+ assert!(
+ config.get_algorithm().is_err(),
+ "{asymmetric} has no key material and must not resolve"
+ );
+ }
+ }
+
#[test]
fn jwt_secrets_are_never_serialized() {
// current_config.toml (and the diagnostic snapshot that cats it) is
diff --git a/core/configs/src/common/mod.rs b/core/configs/src/common/mod.rs
index b14356b0a..4b2257a5d 100644
--- a/core/configs/src/common/mod.rs
+++ b/core/configs/src/common/mod.rs
@@ -19,7 +19,6 @@
//! [`system::SystemConfig`], the HTTP section, and the top-level sections
//! that [`crate::server_config::server::ServerConfig`] composes.
-pub mod cache_indexes;
pub mod defaults;
pub mod displays;
pub mod http;
diff --git a/core/configs/src/common/server.rs b/core/configs/src/common/server.rs
index 214ed085d..315abeadc 100644
--- a/core/configs/src/common/server.rs
+++ b/core/configs/src/common/server.rs
@@ -59,16 +59,6 @@ pub struct MessagesMaintenanceConfig {
pub interval: IggyDuration,
}
-#[serde_as]
-#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
-pub struct MessageSaverConfig {
- pub enabled: bool,
- pub enforce_fsync: bool,
- #[config_env(leaf)]
- #[serde_as(as = "DisplayFromStr")]
- pub interval: IggyDuration,
-}
-
#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct PersonalAccessTokenConfig {
pub max_tokens_per_user: u32,
@@ -99,9 +89,6 @@ pub struct ConsumerGroupConfig {
#[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub rebalancing_timeout: IggyDuration,
- #[config_env(leaf)]
- #[serde_as(as = "DisplayFromStr")]
- pub rebalancing_check_interval: IggyDuration,
}
#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
diff --git a/core/configs/src/common/system.rs b/core/configs/src/common/system.rs
index a9172b0fb..ae0e7eeb2 100644
--- a/core/configs/src/common/system.rs
+++ b/core/configs/src/common/system.rs
@@ -15,11 +15,10 @@
// specific language governing permissions and limitations
// under the License.
-use super::cache_indexes::CacheIndexesConfig;
use super::server::MemoryPoolConfig;
use configs::{ConfigEnv, ConfigEnvMappings};
use iggy_common::IggyByteSize;
-use iggy_common::{CompressionAlgorithm, IggyDuration};
+use iggy_common::IggyDuration;
use serde::{Deserialize, Serialize};
use serde_with::DisplayFromStr;
use serde_with::serde_as;
@@ -35,8 +34,6 @@ pub const LOG_EXTENSION: &str = "log";
#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct SystemConfig<S: ConfigEnvMappings> {
pub path: String,
- pub backup: BackupConfig,
- pub state: StateConfig,
pub runtime: RuntimeConfig,
pub logging: LoggingConfig,
pub stream: StreamConfig,
@@ -44,40 +41,16 @@ pub struct SystemConfig<S: ConfigEnvMappings> {
pub partition: PartitionConfig,
pub segment: SegmentConfig,
pub encryption: EncryptionConfig,
- pub compression: CompressionConfig,
pub recovery: RecoveryConfig,
pub memory_pool: MemoryPoolConfig,
pub sharding: S,
}
-#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
-pub struct BackupConfig {
- pub path: String,
- pub compatibility: CompatibilityConfig,
-}
-
-#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
-pub struct CompatibilityConfig {
- pub path: String,
-}
-
-#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
-pub struct DatabaseConfig {
- pub path: String,
-}
-
#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct RuntimeConfig {
pub path: String,
}
-#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
-pub struct CompressionConfig {
- pub allow_override: bool,
- #[config_env(leaf)]
- pub default_algorithm: CompressionAlgorithm,
-}
-
#[serde_as]
#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct LoggingConfig {
@@ -94,9 +67,6 @@ pub struct LoggingConfig {
#[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub retention: IggyDuration,
- #[config_env(leaf)]
- #[serde_as(as = "DisplayFromStr")]
- pub sysinfo_print_interval: IggyDuration,
}
impl From<&LoggingConfig> for LoggingSettings {
@@ -156,21 +126,9 @@ pub struct RecoveryConfig {
/// `iggy_common::DEFAULT_SEGMENT_SIZE` / `DEFAULT_PREALLOCATE_SEGMENTS`.
#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct SegmentConfig {
- #[config_env(leaf)]
- pub cache_indexes: CacheIndexesConfig,
pub archive_expired: bool,
}
-#[serde_as]
-#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
-pub struct StateConfig {
- pub enforce_fsync: bool,
- pub max_file_operation_retries: u32,
- #[config_env(leaf)]
- #[serde_as(as = "DisplayFromStr")]
- pub retry_delay: IggyDuration,
-}
-
impl<S: ConfigEnvMappings> SystemConfig<S> {
pub fn get_system_path(&self) -> String {
self.path.to_string()
@@ -191,10 +149,6 @@ impl<S: ConfigEnvMappings> SystemConfig<S> {
format!("{}/tokens", self.get_state_path())
}
- pub fn get_backup_path(&self) -> String {
- format!("{}/{}", self.get_system_path(), self.backup.path)
- }
-
pub fn get_runtime_path(&self) -> String {
format!("{}/{}", self.get_system_path(), self.runtime.path)
}
diff --git a/core/configs/src/common/validators.rs b/core/configs/src/common/validators.rs
index 34bfb82a4..de3da58c2 100644
--- a/core/configs/src/common/validators.rs
+++ b/core/configs/src/common/validators.rs
@@ -16,37 +16,19 @@
// under the License.
use super::COMPONENT;
-use super::server::{
- DataMaintenanceConfig, MessageSaverConfig, MessagesMaintenanceConfig, TelemetryConfig,
-};
+use super::server::{DataMaintenanceConfig, MessagesMaintenanceConfig, TelemetryConfig};
use super::server::{MemoryPoolConfig, PersonalAccessTokenConfig};
use super::system::SegmentConfig;
-use super::system::{CompressionConfig, LoggingConfig, PartitionConfig};
+use super::system::{LoggingConfig, PartitionConfig};
use crate::ConfigurationError;
use cpu_allocation::{CpuAllocation, allowed_cpus};
use err_trail::ErrContext;
-use iggy_common::CompressionAlgorithm;
use iggy_common::Validatable;
use std::thread::available_parallelism;
-use tracing::warn;
/// 1 GiB max segment size.
pub const SEGMENT_MAX_SIZE_BYTES: u64 = 1024 * 1024 * 1024;
-impl Validatable<ConfigurationError> for CompressionConfig {
- fn validate(&self) -> Result<(), ConfigurationError> {
- let compression_alg = &self.default_algorithm;
- if *compression_alg != CompressionAlgorithm::None {
- // TODO(numinex): Change this message once server side compression is fully developed.
- warn!(
- "Server started with server-side compression enabled, using algorithm: {compression_alg}, this feature is not implemented yet!"
- );
- }
-
- Ok(())
- }
-}
-
impl Validatable<ConfigurationError> for TelemetryConfig {
fn validate(&self) -> Result<(), ConfigurationError> {
if !self.enabled {
@@ -89,17 +71,6 @@ impl Validatable<ConfigurationError> for SegmentConfig {
}
}
-impl Validatable<ConfigurationError> for MessageSaverConfig {
- fn validate(&self) -> Result<(), ConfigurationError> {
- if self.enabled && self.interval.is_zero() {
- eprintln!("message_saver.interval cannot be zero when message_saver is enabled");
- return Err(ConfigurationError::InvalidConfigurationValue);
- }
-
- Ok(())
- }
-}
-
impl Validatable<ConfigurationError> for DataMaintenanceConfig {
fn validate(&self) -> Result<(), ConfigurationError> {
self.messages.validate().error(|e: &ConfigurationError| {
diff --git a/core/configs/src/lib.rs b/core/configs/src/lib.rs
index 3b2c59541..db8f674b6 100644
--- a/core/configs/src/lib.rs
+++ b/core/configs/src/lib.rs
@@ -20,7 +20,7 @@ extern crate self as configs;
mod common;
mod configs_impl;
mod server_config;
-pub use common::{COMPONENT, cache_indexes, defaults, displays, http, system, validators};
+pub use common::{COMPONENT, defaults, displays, http, system, validators};
pub use configs_derive::ConfigEnv;
pub use configs_impl::{
ConfigEnvMappings, ConfigProvider, ConfigurationError, ConfigurationType, EnvVarMapping,
diff --git a/core/configs/src/server_config/defaults.rs b/core/configs/src/server_config/defaults.rs
index 33422f752..c9659d317 100644
--- a/core/configs/src/server_config/defaults.rs
+++ b/core/configs/src/server_config/defaults.rs
@@ -19,9 +19,9 @@
//! `quic`, `cluster`, `metadata`, `partition`, `message_bus`), sourced
//! from `core/server/config.toml` via [`SERVER_CONFIG`]. Sections drawn
//! from [`crate::common`] (`http`, `system`, `telemetry`,
-//! `consumer_group`, `data_maintenance`, `message_saver`,
-//! `personal_access_token`, `heartbeat`) delegate to the `Default` impls
-//! in [`crate::common::defaults`].
+//! `consumer_group`, `data_maintenance`, `personal_access_token`,
+//! `heartbeat`) delegate to the `Default` impls in
+//! [`crate::common::defaults`].
use super::cluster::{
ClusterAuthConfig, ClusterConfig, ClusterNodeConfig, ClusterTlsConfig, TransportPorts,
@@ -29,15 +29,15 @@ use super::cluster::{
use super::message_bus::MessageBusConfig;
use super::metadata::MetadataConfig;
use super::partition::PartitionConfig;
-use super::quic::{QuicCertificateConfig, QuicConfig, QuicSocketConfig};
+use super::quic::{QuicCertificateConfig, QuicConfig};
use super::server::ServerSystemConfig;
use super::server::{ExtraConfig, ServerConfig};
-use super::tcp::{TcpConfig, TcpSocketConfig, TcpTlsConfig};
+use super::tcp::{TcpConfig, TcpTlsConfig};
use super::websocket::{WebSocketConfig, WebSocketTlsConfig};
use crate::common::http::HttpConfig;
use crate::common::server::{
- ConsumerGroupConfig, DataMaintenanceConfig, HeartbeatConfig, MessageSaverConfig,
- PersonalAccessTokenConfig, TelemetryConfig,
+ ConsumerGroupConfig, DataMaintenanceConfig, HeartbeatConfig, PersonalAccessTokenConfig,
+ TelemetryConfig,
};
use std::sync::Arc;
@@ -52,7 +52,6 @@ impl Default for ServerConfig {
data_maintenance: DataMaintenanceConfig::default(),
extra: ExtraConfig::default(),
heartbeat: HeartbeatConfig::default(),
- message_saver: MessageSaverConfig::default(),
personal_access_token: PersonalAccessTokenConfig::default(),
system: Arc::new(ServerSystemConfig::default()),
quic: QuicConfig::default(),
@@ -186,29 +185,13 @@ impl Default for QuicConfig {
enabled: SERVER_CONFIG.quic.enabled,
address: SERVER_CONFIG.quic.address.parse().unwrap(),
max_concurrent_bidi_streams: SERVER_CONFIG.quic.max_concurrent_bidi_streams as u64,
- datagram_send_buffer_size: SERVER_CONFIG
- .quic
- .datagram_send_buffer_size
- .parse()
- .unwrap(),
initial_mtu: SERVER_CONFIG.quic.initial_mtu.parse().unwrap(),
send_window: SERVER_CONFIG.quic.send_window.parse().unwrap(),
receive_window: SERVER_CONFIG.quic.receive_window.parse().unwrap(),
+ stream_receive_window: SERVER_CONFIG.quic.stream_receive_window.parse().unwrap(),
keep_alive_interval: SERVER_CONFIG.quic.keep_alive_interval.parse().unwrap(),
max_idle_timeout: SERVER_CONFIG.quic.max_idle_timeout.parse().unwrap(),
certificate: QuicCertificateConfig::default(),
- socket: QuicSocketConfig::default(),
- }
- }
-}
-
-impl Default for QuicSocketConfig {
- fn default() -> QuicSocketConfig {
- QuicSocketConfig {
- override_defaults: SERVER_CONFIG.quic.socket.override_defaults,
- recv_buffer_size: SERVER_CONFIG.quic.socket.recv_buffer_size.parse().unwrap(),
- send_buffer_size: SERVER_CONFIG.quic.socket.send_buffer_size.parse().unwrap(),
- keepalive: SERVER_CONFIG.quic.socket.keepalive,
}
}
}
@@ -228,10 +211,7 @@ impl Default for TcpConfig {
TcpConfig {
enabled: SERVER_CONFIG.tcp.enabled,
address: SERVER_CONFIG.tcp.address.parse().unwrap(),
- ipv6: SERVER_CONFIG.tcp.ipv_6,
tls: TcpTlsConfig::default(),
- socket: TcpSocketConfig::default(),
- socket_migration: SERVER_CONFIG.tcp.socket_migration,
}
}
}
@@ -247,19 +227,6 @@ impl Default for TcpTlsConfig {
}
}
-impl Default for TcpSocketConfig {
- fn default() -> TcpSocketConfig {
- TcpSocketConfig {
- override_defaults: SERVER_CONFIG.tcp.socket.override_defaults,
- recv_buffer_size: SERVER_CONFIG.tcp.socket.recv_buffer_size.parse().unwrap(),
- send_buffer_size: SERVER_CONFIG.tcp.socket.send_buffer_size.parse().unwrap(),
- keepalive: SERVER_CONFIG.tcp.socket.keepalive,
- nodelay: SERVER_CONFIG.tcp.socket.nodelay,
- linger: SERVER_CONFIG.tcp.socket.linger.parse().unwrap(),
- }
- }
-}
-
impl Default for WebSocketConfig {
fn default() -> WebSocketConfig {
// The size knobs are optional in the schema (commented-out by
diff --git a/core/configs/src/server_config/displays.rs b/core/configs/src/server_config/displays.rs
index 2fac643c3..642d96970 100644
--- a/core/configs/src/server_config/displays.rs
+++ b/core/configs/src/server_config/displays.rs
@@ -25,22 +25,21 @@
use super::message_bus::MessageBusConfig;
use super::metadata::MetadataConfig;
use super::partition::PartitionConfig;
-use super::quic::{QuicCertificateConfig, QuicConfig, QuicSocketConfig};
+use super::quic::{QuicCertificateConfig, QuicConfig};
use super::server::{ExtraConfig, NamespaceConfig, ServerConfig};
-use super::tcp::{TcpConfig, TcpSocketConfig, TcpTlsConfig};
+use super::tcp::{TcpConfig, TcpTlsConfig};
use std::fmt::{Display, Formatter};
impl Display for ServerConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
- "{{ consumer_group: {}, data_maintenance: {}, extra: {}, message_saver: {}, \
+ "{{ consumer_group: {}, data_maintenance: {}, extra: {}, \
heartbeat: {}, system: {}, quic: {}, tcp: {}, http: {}, telemetry: {}, \
metadata: {}, message_bus: {}, partition: {} }}",
self.consumer_group,
self.data_maintenance,
self.extra,
- self.message_saver,
self.heartbeat,
self.system,
self.quic,
@@ -118,8 +117,8 @@ impl Display for TcpConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
- "{{ enabled: {}, address: {}, ipv6: {}, tls: {}, socket: {}, socket_migration: {} }}",
- self.enabled, self.address, self.ipv6, self.tls, self.socket, self.socket_migration
+ "{{ enabled: {}, address: {}, tls: {} }}",
+ self.enabled, self.address, self.tls
)
}
}
@@ -134,33 +133,18 @@ impl Display for TcpTlsConfig {
}
}
-impl Display for TcpSocketConfig {
- fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
- write!(
- f,
- "{{ override_defaults: {}, recv_buffer_size: {}, send_buffer_size: {}, keepalive: {}, nodelay: {}, linger: {} }}",
- self.override_defaults,
- self.recv_buffer_size,
- self.send_buffer_size,
- self.keepalive,
- self.nodelay,
- self.linger,
- )
- }
-}
-
impl Display for QuicConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
- "{{ enabled: {}, address: {}, max_concurrent_bidi_streams: {}, datagram_send_buffer_size: {}, initial_mtu: {}, send_window: {}, receive_window: {}, keep_alive_interval: {}, max_idle_timeout: {}, certificate: {} }}",
+ "{{ enabled: {}, address: {}, max_concurrent_bidi_streams: {}, initial_mtu: {}, send_window: {}, receive_window: {}, stream_receive_window: {}, keep_alive_interval: {}, max_idle_timeout: {}, certificate: {} }}",
self.enabled,
self.address,
self.max_concurrent_bidi_streams,
- self.datagram_send_buffer_size,
self.initial_mtu,
self.send_window,
self.receive_window,
+ self.stream_receive_window,
self.keep_alive_interval,
self.max_idle_timeout,
self.certificate
@@ -177,13 +161,3 @@ impl Display for QuicCertificateConfig {
)
}
}
-
-impl Display for QuicSocketConfig {
- fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
- write!(
- f,
- "{{ override_defaults: {}, recv_buffer_size: {}, send_buffer_size: {}, keepalive: {} }}",
- self.override_defaults, self.recv_buffer_size, self.send_buffer_size, self.keepalive
- )
- }
-}
diff --git a/core/configs/src/server_config/partition.rs b/core/configs/src/server_config/partition.rs
index 37a593385..d6e327361 100644
--- a/core/configs/src/server_config/partition.rs
+++ b/core/configs/src/server_config/partition.rs
@@ -40,6 +40,7 @@
use super::COMPONENT;
use crate::ConfigurationError;
+use crate::common::validators::SEGMENT_MAX_SIZE_BYTES;
use configs::ConfigEnv;
use iggy_common::{IggyByteSize, Validatable};
use serde::{Deserialize, Serialize};
@@ -61,16 +62,35 @@ pub const DEFAULT_PARTITION_PREPARE_QUEUE_DEPTH: usize = 32;
/// batches) still holds and is looser, so the wire is what decides.
pub const MAX_PARTITION_PREPARE_QUEUE_DEPTH: usize = 127;
+/// Most one segment can overshoot its size cap: rotation checks the cap AFTER
+/// appending, so a sealed segment runs up to one maximum batch past the
+/// per-topic `segment_size`, whose own ceiling is the compile-time
+/// [`SEGMENT_MAX_SIZE_BYTES`]. Tracks the shipped
+/// `message_bus.max_message_size`, which is what actually bounds an appendable
+/// batch.
+///
+/// Mirrors `shard::SEGMENT_SIZE_OVERSHOOT_BYTES`, which the serving side uses as
+/// the floor of its admission divisor.
+pub const SEGMENT_SIZE_OVERSHOOT_BYTES: u64 = 64 * 1024 * 1024;
+
+/// Distinct max-size segments the served-payload budget is sized to hold at
+/// once, and so the concurrent state transfers a shard admits.
+///
+/// Mirrors `shard::CONCURRENT_SERVED_SEGMENTS`.
+pub const CONCURRENT_SERVED_SEGMENTS: u64 = 2;
+
/// Mirrors the free const `shard::PARTITION_ARTIFACT_LEN_DEFAULT` (segment
/// ceiling plus the one whole batch a segment may close past it).
-pub const DEFAULT_TRANSFER_ARTIFACT_BYTES_MAX: u64 = 1024 * 1024 * 1024 + 64 * 1024 * 1024;
+pub const DEFAULT_TRANSFER_ARTIFACT_BYTES_MAX: u64 =
+ SEGMENT_MAX_SIZE_BYTES + SEGMENT_SIZE_OVERSHOOT_BYTES;
/// Mirrors the free const `shard::SERVED_SEGMENT_CACHE_BYTES_DEFAULT`: room for
-/// two concurrently served segments at the size a SEALED one actually reaches,
-/// which is the artifact ceiling above, not the configured segment target. Sized
-/// off the target instead, two admitted pulls would not both fit and would evict
-/// each other on every chunk.
-pub const DEFAULT_TRANSFER_SERVED_CACHE_BYTES_MAX: u64 = 2 * DEFAULT_TRANSFER_ARTIFACT_BYTES_MAX;
+/// [`CONCURRENT_SERVED_SEGMENTS`] served segments at the size a SEALED one
+/// actually reaches, which is the artifact ceiling above, not the configured
+/// segment target. Sized off the target instead, two admitted pulls would not
+/// both fit and would evict each other on every chunk.
+pub const DEFAULT_TRANSFER_SERVED_CACHE_BYTES_MAX: u64 =
+ CONCURRENT_SERVED_SEGMENTS * DEFAULT_TRANSFER_ARTIFACT_BYTES_MAX;
/// Upper bound on the two state-transfer byte knobs. A typo guard, not a sizing
/// endorsement: both are PER SHARD, so a slipped digit multiplies by the core
diff --git a/core/configs/src/server_config/quic.rs b/core/configs/src/server_config/quic.rs
index b6a52f307..614e1002c 100644
--- a/core/configs/src/server_config/quic.rs
+++ b/core/configs/src/server_config/quic.rs
@@ -32,31 +32,20 @@ pub struct QuicConfig {
pub address: String,
pub max_concurrent_bidi_streams: u64,
#[config_env(leaf)]
- pub datagram_send_buffer_size: IggyByteSize,
- #[config_env(leaf)]
pub initial_mtu: IggyByteSize,
#[config_env(leaf)]
pub send_window: IggyByteSize,
#[config_env(leaf)]
pub receive_window: IggyByteSize,
#[config_env(leaf)]
+ pub stream_receive_window: IggyByteSize,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub keep_alive_interval: IggyDuration,
#[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub max_idle_timeout: IggyDuration,
pub certificate: QuicCertificateConfig,
- pub socket: QuicSocketConfig,
-}
-
-#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
-pub struct QuicSocketConfig {
- pub override_defaults: bool,
- #[config_env(leaf)]
- pub recv_buffer_size: IggyByteSize,
- #[config_env(leaf)]
- pub send_buffer_size: IggyByteSize,
- pub keepalive: bool,
}
#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
@@ -85,16 +74,6 @@ impl Validatable<ConfigurationError> for QuicConfig {
);
return Err(ConfigurationError::InvalidConfigurationValue);
}
- // The datagram send buffer is materialized as a `Vec<u8>` of this
- // length in compio-quic, so it must fit in `usize` on the target
- // platform.
- if usize::try_from(self.datagram_send_buffer_size.as_bytes_u64()).is_err() {
- eprintln!(
- "{COMPONENT} quic.datagram_send_buffer_size ({} bytes) does not fit in usize on this target",
- self.datagram_send_buffer_size.as_bytes_u64()
- );
- return Err(ConfigurationError::InvalidConfigurationValue);
- }
// RFC 9000 §14: minimum required MTU is 1200 bytes; quinn stores
// initial_mtu as u16 (max 65535).
let initial_mtu = self.initial_mtu.as_bytes_u64();
@@ -118,6 +97,28 @@ impl Validatable<ConfigurationError> for QuicConfig {
);
return Err(ConfigurationError::InvalidConfigurationValue);
}
+ // Per-stream window: nonzero (a zero window blocks every stream
+ // forever), u32 for quinn's VarInt, and no larger than the
+ // connection window it is carved from - the connection window caps
+ // every stream, so the excess could never be used.
+ let stream_receive_window = self.stream_receive_window.as_bytes_u64();
+ if stream_receive_window == 0 {
+ eprintln!("{COMPONENT} quic.stream_receive_window must be > 0");
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if u32::try_from(stream_receive_window).is_err() {
+ eprintln!(
+ "{COMPONENT} quic.stream_receive_window ({stream_receive_window} bytes) does not fit in u32 (quinn VarInt limit)",
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if stream_receive_window > self.receive_window.as_bytes_u64() {
+ eprintln!(
+ "{COMPONENT} quic.stream_receive_window ({stream_receive_window} bytes) exceeds quic.receive_window ({} bytes)",
+ self.receive_window.as_bytes_u64()
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
// `send_window` is u64-sized in QuicTuning, but quinn's VarInt
// protocol-level cap is 2^62 - 1; reject anything above that.
const QUINN_VARINT_MAX: u64 = (1u64 << 62) - 1;
@@ -128,6 +129,16 @@ impl Validatable<ConfigurationError> for QuicConfig {
);
return Err(ConfigurationError::InvalidConfigurationValue);
}
+ // quinn's IdleTimeout is a VarInt count of MILLISECONDS, so the
+ // bound bites long before an IggyDuration overflows. Zero means
+ // "disabled" and never reaches the conversion.
+ let max_idle_timeout_ms = self.max_idle_timeout.get_duration().as_millis();
+ if max_idle_timeout_ms > u128::from(QUINN_VARINT_MAX) {
+ eprintln!(
+ "{COMPONENT} quic.max_idle_timeout ({max_idle_timeout_ms} ms) exceeds quinn VarInt max (2^62 - 1)",
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
Ok(())
}
}
@@ -141,10 +152,10 @@ mod tests {
enabled: false,
address: String::new(),
max_concurrent_bidi_streams: 1,
- datagram_send_buffer_size: IggyByteSize::from(100_u64 * 1024),
- initial_mtu: IggyByteSize::from(8_u64 * 1024),
+ initial_mtu: IggyByteSize::from(1200_u64),
send_window: IggyByteSize::from(64_u64 * 1024 * 1024),
receive_window: IggyByteSize::from(64_u64 * 1024 * 1024),
+ stream_receive_window: IggyByteSize::from(16_u64 * 1024 * 1024),
keep_alive_interval: IggyDuration::from(std::time::Duration::from_secs(10)),
max_idle_timeout: IggyDuration::from(std::time::Duration::from_secs(30)),
certificate: QuicCertificateConfig {
@@ -152,12 +163,6 @@ mod tests {
cert_file: String::new(),
key_file: String::new(),
},
- socket: QuicSocketConfig {
- override_defaults: false,
- recv_buffer_size: IggyByteSize::from(0_u64),
- send_buffer_size: IggyByteSize::from(0_u64),
- keepalive: false,
- },
}
}
@@ -214,4 +219,47 @@ mod tests {
c.initial_mtu = IggyByteSize::from(1200_u64);
assert!(c.validate().is_ok());
}
+
+ #[test]
+ fn rejects_zero_stream_receive_window() {
+ let mut c = baseline();
+ c.stream_receive_window = IggyByteSize::from(0_u64);
+ assert!(c.validate().is_err());
+ }
+
+ #[test]
+ fn rejects_stream_receive_window_above_u32() {
+ let mut c = baseline();
+ c.receive_window = IggyByteSize::from(u64::from(u32::MAX));
+ c.stream_receive_window = IggyByteSize::from(u64::from(u32::MAX) + 1);
+ assert!(c.validate().is_err());
+ }
+
+ #[test]
+ fn rejects_stream_receive_window_above_receive_window() {
+ let mut c = baseline();
+ c.stream_receive_window = IggyByteSize::from(65_u64 * 1024 * 1024);
+ assert!(c.validate().is_err());
+ }
+
+ #[test]
+ fn accepts_stream_receive_window_equal_to_receive_window() {
+ let mut c = baseline();
+ c.stream_receive_window = c.receive_window;
+ assert!(c.validate().is_ok());
+ }
+
+ #[test]
+ fn rejects_max_idle_timeout_above_quinn_varint_max() {
+ let mut c = baseline();
+ c.max_idle_timeout = IggyDuration::from(std::time::Duration::from_millis(1u64 << 62));
+ assert!(c.validate().is_err());
+ }
+
+ #[test]
+ fn accepts_max_idle_timeout_at_quinn_varint_max() {
+ let mut c = baseline();
+ c.max_idle_timeout = IggyDuration::from(std::time::Duration::from_millis((1u64 << 62) - 1));
+ assert!(c.validate().is_ok());
+ }
}
diff --git a/core/configs/src/server_config/server.rs b/core/configs/src/server_config/server.rs
index 5f0522c1a..f97b4b903 100644
--- a/core/configs/src/server_config/server.rs
+++ b/core/configs/src/server_config/server.rs
@@ -42,9 +42,8 @@ use std::sync::Arc;
pub use crate::common::server::{
ConsumerGroupConfig, DataMaintenanceConfig, HeartbeatConfig, MemoryPoolConfig,
- MessageSaverConfig, MessagesMaintenanceConfig, PersonalAccessTokenCleanerConfig,
- PersonalAccessTokenConfig, TelemetryConfig, TelemetryLogsConfig, TelemetryTracesConfig,
- TelemetryTransport,
+ MessagesMaintenanceConfig, PersonalAccessTokenCleanerConfig, PersonalAccessTokenConfig,
+ TelemetryConfig, TelemetryLogsConfig, TelemetryTracesConfig, TelemetryTransport,
};
const DEFAULT_CONFIG_PATH: &str = "core/server/config.toml";
@@ -108,7 +107,6 @@ pub struct ServerConfig {
pub data_maintenance: DataMaintenanceConfig,
#[serde(default)]
pub extra: ExtraConfig,
- pub message_saver: MessageSaverConfig,
pub personal_access_token: PersonalAccessTokenConfig,
pub heartbeat: HeartbeatConfig,
pub system: Arc<ServerSystemConfig>,
diff --git a/core/configs/src/server_config/tcp.rs b/core/configs/src/server_config/tcp.rs
index 8e0acafdb..066218c3c 100644
--- a/core/configs/src/server_config/tcp.rs
+++ b/core/configs/src/server_config/tcp.rs
@@ -18,19 +18,13 @@
//! TCP listener schema.
use configs::ConfigEnv;
-use iggy_common::{IggyByteSize, IggyDuration};
use serde::{Deserialize, Serialize};
-use serde_with::DisplayFromStr;
-use serde_with::serde_as;
#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct TcpConfig {
pub enabled: bool,
pub address: String,
- pub ipv6: bool,
pub tls: TcpTlsConfig,
- pub socket: TcpSocketConfig,
- pub socket_migration: bool,
}
#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
@@ -40,18 +34,3 @@ pub struct TcpTlsConfig {
pub cert_file: String,
pub key_file: String,
}
-
-#[serde_as]
-#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
-pub struct TcpSocketConfig {
- pub override_defaults: bool,
- #[config_env(leaf)]
- pub recv_buffer_size: IggyByteSize,
- #[config_env(leaf)]
- pub send_buffer_size: IggyByteSize,
- pub keepalive: bool,
- pub nodelay: bool,
- #[config_env(leaf)]
- #[serde_as(as = "DisplayFromStr")]
- pub linger: IggyDuration,
-}
diff --git a/core/configs/src/server_config/validators.rs b/core/configs/src/server_config/validators.rs
index 60541d449..e3c6b0d99 100644
--- a/core/configs/src/server_config/validators.rs
+++ b/core/configs/src/server_config/validators.rs
@@ -19,18 +19,19 @@
//!
//! Delegates section by section, including
//! [`super::message_bus::MessageBusConfig::validate`], then applies the
-//! cross-section invariants: topic vs segment sizing, JWT gating when
-//! HTTP is enabled, and server-default expiry sanity.
+//! cross-section invariants: JWT gating when HTTP is enabled and
+//! server-default expiry sanity.
use super::COMPONENT;
use super::cluster::STATE_CHUNK_HEADER_LEN;
+use super::partition::{CONCURRENT_SERVED_SEGMENTS, SEGMENT_SIZE_OVERSHOOT_BYTES};
use super::server::{ExtraConfig, NamespaceConfig, ServerConfig};
use crate::ConfigurationError;
+use crate::common::http::HMAC_JWT_ALGORITHMS;
use crate::common::validators::SEGMENT_MAX_SIZE_BYTES;
use err_trail::ErrContext;
use iggy_common::{IggyExpiry, Validatable};
use server_common::sharding::IggyNamespace;
-use tracing::warn;
/// compio-ws (tungstenite 0.29) `write_buffer_size` default. Used to
/// evaluate the `max_write_buffer_size > write_buffer_size` invariant
@@ -67,12 +68,6 @@ impl Validatable<ConfigurationError> for ServerConfig {
.error(|e: &ConfigurationError| {
format!("{COMPONENT} (error: {e}) - failed to validate segment config")
})?;
- self.system
- .compression
- .validate()
- .error(|e: &ConfigurationError| {
- format!("{COMPONENT} (error: {e}) - failed to validate compression config")
- })?;
self.telemetry.validate().error(|e: &ConfigurationError| {
format!("{COMPONENT} (error: {e}) - failed to validate telemetry config")
})?;
@@ -97,11 +92,6 @@ impl Validatable<ConfigurationError> for ServerConfig {
.error(|e: &ConfigurationError| {
format!("{COMPONENT} (error: {e}) - failed to validate logging config")
})?;
- self.message_saver
- .validate()
- .error(|e: &ConfigurationError| {
- format!("{COMPONENT} (error: {e}) - failed to validate message saver config")
- })?;
if self.http.enabled
&& let IggyExpiry::ServerDefault = self.http.jwt.access_token_expiry
@@ -110,6 +100,24 @@ impl Validatable<ConfigurationError> for ServerConfig {
return Err(ConfigurationError::InvalidConfigurationValue);
}
+ // The signing key is always HMAC (built from a shared secret; there is
+ // no PEM loading path), and jsonwebtoken refuses a header algorithm
+ // whose family does not match the key. An asymmetric algorithm builds
+ // the manager fine and then fails every login at token generation, so
+ // the server would serve HTTP that nobody can authenticate against.
+ if self.http.enabled && self.http.jwt.get_algorithm().is_err() {
+ let supported = HMAC_JWT_ALGORITHMS.map(|(name, _)| name).join(", ");
+ eprintln!(
+ "http.jwt.algorithm '{}' is not supported: the server signs with a shared \
+ secret, so only {supported} can be used. This field governs self-issued \
+ tokens only; [[http.jwt.trusted_issuers]] is unaffected, because an issuer's \
+ token is verified with the algorithm named in its own header and the matching \
+ JWKS key, never with this field",
+ self.http.jwt.algorithm
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
if self.http.enabled
&& self.http.tls.enabled
&& (self.http.tls.cert_file.is_empty() || self.http.tls.key_file.is_empty())
@@ -172,6 +180,32 @@ impl Validatable<ConfigurationError> for ServerConfig {
return Err(ConfigurationError::InvalidConfigurationValue);
}
+ // Raising the artifact ceiling (or the segment size) without raising the
+ // served-payload budget silently serialises rejoins: a whole-node rejoin
+ // then walks its partitions one at a time. Warn rather than reject,
+ // because running under the budget costs re-reads, not failures, and an
+ // operator deliberately trading serving concurrency for memory still
+ // boots.
+ let (resident_len, served_slots) = served_transfer_slots(self);
+ if served_slots < CONCURRENT_SERVED_SEGMENTS {
+ let served_cache = self
+ .partition
+ .transfer_served_cache_bytes_max
+ .as_bytes_u64();
+ eprintln!(
+ "{COMPONENT} partition.transfer_served_cache_bytes_max ({served_cache} B) holds \
+ only {served_slots} state-transfer slot(s) of {resident_len} B (the larger of \
+ partition.transfer_artifact_bytes_max and the built-in segment ceiling of \
+ {SEGMENT_MAX_SIZE_BYTES} B + {SEGMENT_SIZE_OVERSHOOT_BYTES} B of overshoot), \
+ below the {CONCURRENT_SERVED_SEGMENTS} this shard is sized to serve at once, so \
+ rejoins serialise and every cache miss re-reads and re-hashes a whole segment to \
+ serve one chunk. The segment ceiling is a compile-time constant, so the only \
+ knobs here are these two: raise transfer_served_cache_bytes_max to {} B, or \
+ lower transfer_artifact_bytes_max",
+ resident_len.saturating_mul(CONCURRENT_SERVED_SEGMENTS)
+ );
+ }
+
self.message_bus
.validate()
.error(|e: &ConfigurationError| {
@@ -313,20 +347,48 @@ impl Validatable<ConfigurationError> for ServerConfig {
return Err(ConfigurationError::InvalidConfigurationValue);
}
- reject_unsupported_and_warn_inert(self)?;
+ reject_unsupported(self)?;
Ok(())
}
}
+/// `(resident_len, slots)`: the size a SEALED segment actually reaches, and how
+/// many of them the served-payload budget holds at once.
+///
+/// Reproduces what `IggyShard::partition_transfer_admission_cap` computes at the
+/// shipped defaults, which is what really decides how many rejoins one shard
+/// serves concurrently. The shard divides by the CONFIGURED segment size plus
+/// overshoot; this divides by the compile-time segment ceiling, and the two
+/// coincide only because bootstrap pins the deployed segment size to
+/// `DEFAULT_SEGMENT_SIZE`. Where they diverge this side is the conservative one,
+/// so the warning it drives can over-fire but never under-fire.
+///
+/// The divisor is the sealed size rather than the configured target because
+/// rotation checks the cap after appending. Segment size is per topic now, so
+/// the sealed bound is the LARGEST segment any topic may legally declare, not a
+/// configured value. The `max` floor is never zero, so the division always
+/// holds, and the quotient carries the shard's own one-slot clamp so the warning
+/// never claims a slot count the runtime would not admit.
+fn served_transfer_slots(config: &ServerConfig) -> (u64, u64) {
+ let resident_len = config
+ .partition
+ .transfer_artifact_bytes_max
+ .as_bytes_u64()
+ .max(SEGMENT_MAX_SIZE_BYTES.saturating_add(SEGMENT_SIZE_OVERSHOOT_BYTES));
+ let slots = (config
+ .partition
+ .transfer_served_cache_bytes_max
+ .as_bytes_u64()
+ / resident_len)
+ .max(1);
+ (resident_len, slots)
+}
+
/// The server parses the whole config surface but does not yet honor every
-/// knob. Make the still-inert ones loud at boot: reject the unsupported
-/// features (all off by default, so only a deliberate opt-in trips this) and
-/// warn once for tuning knobs the server silently ignores. Warnings fire only
-/// when a knob deviates from its [`ServerConfig::default`] baseline, so a
-/// pristine config.toml boots without noise. The guard test below pins the
-/// compared knobs against drift.
-fn reject_unsupported_and_warn_inert(config: &ServerConfig) -> Result<(), ConfigurationError> {
+/// knob. Make the still-inert ones loud at boot rather than silently ignored.
+/// All are off by default, so only a deliberate opt-in trips this.
+fn reject_unsupported(config: &ServerConfig) -> Result<(), ConfigurationError> {
if config.system.segment.archive_expired {
eprintln!("system.segment.archive_expired is not supported");
return Err(ConfigurationError::InvalidConfigurationValue);
@@ -336,62 +398,6 @@ fn reject_unsupported_and_warn_inert(config: &ServerConfig) -> Result<(), Config
return Err(ConfigurationError::InvalidConfigurationValue);
}
- let defaults = ServerConfig::default();
-
- if config.tcp.socket.override_defaults {
- warn!("tcp.socket tuning is set but not applied");
- }
- if config.quic.socket.override_defaults {
- warn!("quic.socket tuning is set but not applied");
- }
- if config.tcp.ipv6 {
- warn!("tcp.ipv6 is ignored; IPv4 vs IPv6 is decided by the tcp.address string");
- }
- if config.tcp.socket_migration != defaults.tcp.socket_migration {
- warn!("tcp.socket_migration is not implemented");
- }
- if config.system.segment.cache_indexes != defaults.system.segment.cache_indexes {
- warn!("system.segment.cache_indexes is not applied");
- }
- if config.system.logging.sysinfo_print_interval
- != defaults.system.logging.sysinfo_print_interval
- {
- warn!("system.logging.sysinfo_print_interval is not applied");
- }
- if config.system.backup.path != defaults.system.backup.path
- || config.system.backup.compatibility.path != defaults.system.backup.compatibility.path
- {
- warn!("backup is not supported");
- }
- // default_algorithm deviation is already warned by the delegated legacy
- // CompressionConfig::validate; only allow_override needs a signal here.
- if config.system.compression.allow_override != defaults.system.compression.allow_override {
- warn!(
- "system.compression.allow_override is inert; live compression is per-topic from the request"
- );
- }
- if config.system.state.enforce_fsync != defaults.system.state.enforce_fsync
- || config.system.state.max_file_operation_retries
- != defaults.system.state.max_file_operation_retries
- || config.system.state.retry_delay != defaults.system.state.retry_delay
- {
- warn!(
- "system.state tuning (enforce_fsync, max_file_operation_retries, retry_delay) is not applied"
- );
- }
- if config.consumer_group.rebalancing_check_interval
- != defaults.consumer_group.rebalancing_check_interval
- {
- warn!(
- "consumer_group.rebalancing_check_interval is not applied; rebalancing cadence uses system.sharding.reconcile_periodic_interval"
- );
- }
- if config.message_saver.interval != defaults.message_saver.interval
- || config.message_saver.enforce_fsync != defaults.message_saver.enforce_fsync
- {
- warn!("periodic message_saver is not implemented; only shutdown-flush is active");
- }
-
Ok(())
}
@@ -490,6 +496,74 @@ mod tests {
assert!(config.validate().is_err());
}
+ #[test]
+ fn given_asymmetric_jwt_algorithm_when_validating_should_reject() {
+ // Boots clean today and then fails every login: the key is HMAC and
+ // jsonwebtoken refuses a mismatched algorithm family at sign time.
+ for algorithm in ["RS256", "RS384", "RS512", "ES256", "not-an-algorithm"] {
+ let config =
+ config_with_override(&format!("[http.jwt]\nalgorithm = \"{algorithm}\"\n"));
+ assert!(
+ config.validate().is_err(),
+ "{algorithm} has no key material and must not boot"
+ );
+ }
+ }
+
+ #[test]
+ fn given_hmac_jwt_algorithm_when_validating_should_pass() {
+ for (algorithm, _) in HMAC_JWT_ALGORITHMS {
+ let config =
+ config_with_override(&format!("[http.jwt]\nalgorithm = \"{algorithm}\"\n"));
+ config
+ .validate()
+ .unwrap_or_else(|_| panic!("{algorithm} is signable and must validate"));
+ }
+ }
+
+ #[test]
+ fn given_asymmetric_jwt_algorithm_when_http_disabled_should_pass() {
+ // No listener means no token is ever issued, so a stale algorithm is
+ // inert rather than a broken deployment.
+ let config =
+ config_with_override("[http]\nenabled = false\n\n[http.jwt]\nalgorithm = \"RS256\"\n");
+ config
+ .validate()
+ .expect("a disabled HTTP listener never signs a token");
+ }
+
+ // The shipped budget is exactly two artifact ceilings, so the pristine
+ // config sits on the boundary and a single MiB past it starves a slot.
+ #[test]
+ fn given_artifact_ceiling_starving_transfer_slots_when_validating_should_warn_only() {
+ let config =
+ config_with_override("[partition]\ntransfer_artifact_bytes_max = \"1089 MiB\"\n");
+ assert_eq!(served_transfer_slots(&config).1, 1);
+ config
+ .validate()
+ .expect("a starved served cache warns but still boots");
+ }
+
+ #[test]
+ fn given_artifact_ceiling_at_transfer_slot_boundary_when_sizing_should_keep_full_slot_count() {
+ let config = config_with_override("");
+ assert_eq!(
+ served_transfer_slots(&config).1,
+ CONCURRENT_SERVED_SEGMENTS,
+ "the shipped defaults must serve the concurrency they are sized for"
+ );
+ }
+
+ #[test]
+ fn given_served_cache_raised_with_artifact_ceiling_when_validating_should_keep_transfer_slots()
+ {
+ let config = config_with_override(
+ "[partition]\ntransfer_artifact_bytes_max = \"2 GiB\"\ntransfer_served_cache_bytes_max = \"4 GiB\"\n",
+ );
+ assert_eq!(served_transfer_slots(&config).1, CONCURRENT_SERVED_SEGMENTS);
+ config.validate().expect("budget raised in step must boot");
+ }
+
#[test]
fn given_peer_queue_capacity_not_above_repair_chunk_max_when_validating_should_reject() {
// The default repair_chunk_max (128) must stay strictly below
@@ -641,62 +715,6 @@ mod tests {
.expect("a disabled heartbeat never reads its interval");
}
- /// The warn-helper baseline is [`ServerConfig::default`], but the reused
- /// legacy sections source that default from the legacy server config.toml,
- /// not this NG file. Pin the knobs the helper compares so any drift between
- /// the two config.toml files fails here instead of as a spurious boot warn.
- #[test]
- fn given_shipped_ng_config_when_compared_to_default_should_match_warned_knobs() {
- let shipped: ServerConfig = Figment::new()
- .merge(Toml::string(DEFAULT_CONFIG))
- .extract()
- .expect("default config deserializes");
- let defaults = ServerConfig::default();
-
- assert_eq!(shipped.tcp.socket_migration, defaults.tcp.socket_migration);
- assert_eq!(
- shipped.system.segment.cache_indexes,
- defaults.system.segment.cache_indexes
- );
- assert_eq!(
- shipped.system.logging.sysinfo_print_interval,
- defaults.system.logging.sysinfo_print_interval
- );
- assert_eq!(shipped.system.backup.path, defaults.system.backup.path);
- assert_eq!(
- shipped.system.backup.compatibility.path,
- defaults.system.backup.compatibility.path
- );
- assert_eq!(
- shipped.system.compression.allow_override,
- defaults.system.compression.allow_override
- );
- assert_eq!(
- shipped.system.state.enforce_fsync,
- defaults.system.state.enforce_fsync
- );
- assert_eq!(
- shipped.system.state.max_file_operation_retries,
- defaults.system.state.max_file_operation_retries
- );
- assert_eq!(
- shipped.system.state.retry_delay,
- defaults.system.state.retry_delay
- );
- assert_eq!(
- shipped.consumer_group.rebalancing_check_interval,
- defaults.consumer_group.rebalancing_check_interval
- );
- assert_eq!(
- shipped.message_saver.interval,
- defaults.message_saver.interval
- );
- assert_eq!(
- shipped.message_saver.enforce_fsync,
- defaults.message_saver.enforce_fsync
- );
- }
-
// http.enabled needs a non-ServerDefault JWT expiry to clear the sibling
// check above; ServerConfig::default() already satisfies that.
fn https_config(cert_file: &str, key_file: &str) -> ServerConfig {
diff --git a/core/connectors/runtime/Cargo.toml b/core/connectors/runtime/Cargo.toml
index 8f73799b0..448c1a404 100644
--- a/core/connectors/runtime/Cargo.toml
+++ b/core/connectors/runtime/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-connectors"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
description = "Connectors runtime for Iggy message streaming platform"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sdk/Cargo.toml b/core/connectors/sdk/Cargo.toml
index 74b6e7d09..bfd99e2fb 100644
--- a/core/connectors/sdk/Cargo.toml
+++ b/core/connectors/sdk/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_sdk"
-version = "0.4.0-edge.1"
+version = "0.4.0-edge.2"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/clickhouse_sink/Cargo.toml b/core/connectors/sinks/clickhouse_sink/Cargo.toml
index c2ae22f45..a624d1cca 100644
--- a/core/connectors/sinks/clickhouse_sink/Cargo.toml
+++ b/core/connectors/sinks/clickhouse_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_clickhouse_sink"
-version = "0.2.0-edge.2"
+version = "0.2.0-edge.3"
description = "Iggy ClickHouse sink connector for streaming messages into ClickHouse"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/delta_sink/Cargo.toml b/core/connectors/sinks/delta_sink/Cargo.toml
index 6a5ec4b60..29b836acd 100644
--- a/core/connectors/sinks/delta_sink/Cargo.toml
+++ b/core/connectors/sinks/delta_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_delta_sink"
-version = "0.2.0-edge.2"
+version = "0.2.0-edge.3"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/doris_sink/Cargo.toml b/core/connectors/sinks/doris_sink/Cargo.toml
index 815cea2b1..a9f228a56 100644
--- a/core/connectors/sinks/doris_sink/Cargo.toml
+++ b/core/connectors/sinks/doris_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_doris_sink"
-version = "0.2.0-edge.2"
+version = "0.2.0-edge.3"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/elasticsearch_sink/Cargo.toml b/core/connectors/sinks/elasticsearch_sink/Cargo.toml
index 0103476fb..3f104c017 100644
--- a/core/connectors/sinks/elasticsearch_sink/Cargo.toml
+++ b/core/connectors/sinks/elasticsearch_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_elasticsearch_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
description = "Iggy Elasticsearch sink connector"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/http_sink/Cargo.toml b/core/connectors/sinks/http_sink/Cargo.toml
index ece790c5a..5ad0d8f48 100644
--- a/core/connectors/sinks/http_sink/Cargo.toml
+++ b/core/connectors/sinks/http_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_http_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
description = "Iggy HTTP sink connector for delivering stream messages to any HTTP endpoint via webhooks, REST APIs, or serverless functions."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/iceberg_sink/Cargo.toml b/core/connectors/sinks/iceberg_sink/Cargo.toml
index 73df01696..14433a39f 100644
--- a/core/connectors/sinks/iceberg_sink/Cargo.toml
+++ b/core/connectors/sinks/iceberg_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_iceberg_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
edition = "2024"
license = "Apache-2.0"
keywords = ["iggy", "messaging", "streaming"]
diff --git a/core/connectors/sinks/influxdb_sink/Cargo.toml b/core/connectors/sinks/influxdb_sink/Cargo.toml
index 4e56e00e4..39e29c729 100644
--- a/core/connectors/sinks/influxdb_sink/Cargo.toml
+++ b/core/connectors/sinks/influxdb_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_influxdb_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
description = "Iggy InfluxDB sink connector for storing stream messages as line protocol"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/meilisearch_sink/Cargo.toml b/core/connectors/sinks/meilisearch_sink/Cargo.toml
index bbc771a7b..a21988a9c 100644
--- a/core/connectors/sinks/meilisearch_sink/Cargo.toml
+++ b/core/connectors/sinks/meilisearch_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_meilisearch_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
description = "Iggy Meilisearch sink connector"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/mongodb_sink/Cargo.toml b/core/connectors/sinks/mongodb_sink/Cargo.toml
index 2ec58fd30..bfe2b5da3 100644
--- a/core/connectors/sinks/mongodb_sink/Cargo.toml
+++ b/core/connectors/sinks/mongodb_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_mongodb_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
description = "Iggy MongoDB sink connector for storing stream messages into MongoDB database"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/postgres_sink/Cargo.toml b/core/connectors/sinks/postgres_sink/Cargo.toml
index e17653b06..374911bc0 100644
--- a/core/connectors/sinks/postgres_sink/Cargo.toml
+++ b/core/connectors/sinks/postgres_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_postgres_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
description = "Iggy PostgreSQL sink connector for storing stream messages into PostgreSQL database"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/quickwit_sink/Cargo.toml b/core/connectors/sinks/quickwit_sink/Cargo.toml
index 71fc393a4..3cca1120d 100644
--- a/core/connectors/sinks/quickwit_sink/Cargo.toml
+++ b/core/connectors/sinks/quickwit_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_quickwit_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/s3_sink/Cargo.toml b/core/connectors/sinks/s3_sink/Cargo.toml
index 484d311f4..150243977 100644
--- a/core/connectors/sinks/s3_sink/Cargo.toml
+++ b/core/connectors/sinks/s3_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_s3_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
description = "Iggy S3 sink connector for writing stream messages to Amazon S3 and S3-compatible stores"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/stdout_sink/Cargo.toml b/core/connectors/sinks/stdout_sink/Cargo.toml
index 3a8730170..f79d5c4b1 100644
--- a/core/connectors/sinks/stdout_sink/Cargo.toml
+++ b/core/connectors/sinks/stdout_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_stdout_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/surrealdb_sink/Cargo.toml b/core/connectors/sinks/surrealdb_sink/Cargo.toml
index 76b301714..ed6a2d0a9 100644
--- a/core/connectors/sinks/surrealdb_sink/Cargo.toml
+++ b/core/connectors/sinks/surrealdb_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_surrealdb_sink"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
description = "Iggy SurrealDB sink connector for writing stream messages into SurrealDB"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sources/elasticsearch_source/Cargo.toml b/core/connectors/sources/elasticsearch_source/Cargo.toml
index 17b5f80cb..75cddf6fe 100644
--- a/core/connectors/sources/elasticsearch_source/Cargo.toml
+++ b/core/connectors/sources/elasticsearch_source/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_elasticsearch_source"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
description = "Iggy Elasticsearch source connector"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sources/influxdb_source/Cargo.toml b/core/connectors/sources/influxdb_source/Cargo.toml
index 173a4c074..3da5c3655 100644
--- a/core/connectors/sources/influxdb_source/Cargo.toml
+++ b/core/connectors/sources/influxdb_source/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_influxdb_source"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
description = "Iggy InfluxDB source connector for polling Flux query results"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sources/postgres_source/Cargo.toml b/core/connectors/sources/postgres_source/Cargo.toml
index 6ebe69b58..48453a805 100644
--- a/core/connectors/sources/postgres_source/Cargo.toml
+++ b/core/connectors/sources/postgres_source/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_postgres_source"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
description = "Iggy PostgreSQL source connector supporting CDC and table polling for message streaming platform"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sources/random_source/Cargo.toml b/core/connectors/sources/random_source/Cargo.toml
index 93ac41587..e3e07c1a6 100644
--- a/core/connectors/sources/random_source/Cargo.toml
+++ b/core/connectors/sources/random_source/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_random_source"
-version = "0.5.0-edge.2"
+version = "0.5.0-edge.3"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/harness_derive/src/attrs.rs b/core/harness_derive/src/attrs.rs
index 76fee823c..9d1f07cbd 100644
--- a/core/harness_derive/src/attrs.rs
+++ b/core/harness_derive/src/attrs.rs
@@ -371,7 +371,7 @@ fn parse_transport_ident(ident: Ident) -> syn::Result<Transport> {
}
}
-/// Parses a dot-notation config key like `segment.cache_indexes` or `metadata.journal_slots`.
+/// Parses a dot-notation config key like `heartbeat.interval` or `metadata.journal_slots`.
fn parse_config_key(input: ParseStream) -> syn::Result<(String, Span)> {
let first: Ident = input.parse()?;
let span = first.span();
@@ -629,9 +629,9 @@ mod tests {
#[test]
fn parse_server_static() {
- let attrs: IggyTestAttrs = syn::parse_quote!(server(segment.cache_indexes = "all"));
- let cache_indexes = attrs.server.find_override("segment.cache_indexes").unwrap();
- assert!(matches!(&cache_indexes.value, ConfigValue::Static(s) if s == "all"));
+ let attrs: IggyTestAttrs = syn::parse_quote!(server(heartbeat.interval = "30s"));
+ let interval = attrs.server.find_override("heartbeat.interval").unwrap();
+ assert!(matches!(&interval.value, ConfigValue::Static(s) if s == "30s"));
}
#[test]
@@ -642,10 +642,9 @@ mod tests {
#[test]
fn parse_server_matrix() {
- let attrs: IggyTestAttrs =
- syn::parse_quote!(server(segment.cache_indexes = ["none", "all"]));
- let cache_indexes = attrs.server.find_override("segment.cache_indexes").unwrap();
- assert!(matches!(&cache_indexes.value, ConfigValue::Matrix(v) if v.len() == 2));
+ let attrs: IggyTestAttrs = syn::parse_quote!(server(heartbeat.interval = ["30s", "60s"]));
+ let interval = attrs.server.find_override("heartbeat.interval").unwrap();
+ assert!(matches!(&interval.value, ConfigValue::Matrix(v) if v.len() == 2));
}
#[test]
@@ -653,21 +652,21 @@ mod tests {
let attrs: IggyTestAttrs = syn::parse_quote!(
test_client_transport = [Tcp, Http],
server(
- segment.cache_indexes = ["none", "all"],
+ heartbeat.interval = ["30s", "60s"],
metadata.journal_slots = 1024,
- tcp.socket.nodelay = true
+ tcp.enabled = true
)
);
assert_eq!(attrs.transports.len(), 2);
- let cache_indexes = attrs.server.find_override("segment.cache_indexes").unwrap();
+ let interval = attrs.server.find_override("heartbeat.interval").unwrap();
let journal_slots = attrs
.server
.find_override("metadata.journal_slots")
.unwrap();
- let tcp_nodelay = attrs.server.find_override("tcp.socket.nodelay").unwrap();
- assert!(matches!(&cache_indexes.value, ConfigValue::Matrix(v) if v.len() == 2));
+ let tcp_enabled = attrs.server.find_override("tcp.enabled").unwrap();
+ assert!(matches!(&interval.value, ConfigValue::Matrix(v) if v.len() == 2));
assert!(matches!(&journal_slots.value, ConfigValue::Static(s) if s == "1024"));
- assert!(matches!(&tcp_nodelay.value, ConfigValue::Static(s) if s == "true"));
+ assert!(matches!(&tcp_enabled.value, ConfigValue::Static(s) if s == "true"));
}
#[test]
@@ -722,11 +721,11 @@ mod tests {
#[test]
fn parse_mcp_combined() {
let attrs: IggyTestAttrs =
- syn::parse_quote!(seed = my_seed, server(mcp, segment.cache_indexes = "all"));
+ syn::parse_quote!(seed = my_seed, server(mcp, heartbeat.interval = "30s"));
assert!(attrs.server.mcp.is_some());
assert!(attrs.seed_fn.is_some());
- let cache_indexes = attrs.server.find_override("segment.cache_indexes").unwrap();
- assert!(matches!(&cache_indexes.value, ConfigValue::Static(s) if s == "all"));
+ let interval = attrs.server.find_override("heartbeat.interval").unwrap();
+ assert!(matches!(&interval.value, ConfigValue::Static(s) if s == "30s"));
}
#[test]
@@ -812,11 +811,11 @@ mod tests {
let attrs: IggyTestAttrs = syn::parse_quote!(
cluster_nodes = [3, 5],
test_client_transport = [Tcp, Http],
- server(segment.cache_indexes = ["none", "all"])
+ server(heartbeat.interval = ["30s", "60s"])
);
assert!(matches!(&attrs.cluster_nodes, ClusterNodesValue::Matrix(v) if v == &[3, 5]));
assert_eq!(attrs.transports.len(), 2);
- let cache_indexes = attrs.server.find_override("segment.cache_indexes").unwrap();
- assert!(matches!(&cache_indexes.value, ConfigValue::Matrix(v) if v.len() == 2));
+ let interval = attrs.server.find_override("heartbeat.interval").unwrap();
+ assert!(matches!(&interval.value, ConfigValue::Matrix(v) if v.len() == 2));
}
}
diff --git a/core/harness_derive/src/codegen.rs b/core/harness_derive/src/codegen.rs
index d6adb360a..481e39af4 100644
--- a/core/harness_derive/src/codegen.rs
+++ b/core/harness_derive/src/codegen.rs
@@ -753,7 +753,7 @@ mod tests {
transport: Transport::Http,
transport_explicit: true,
config_values: vec![
- ("segment.cache_indexes".to_string(), "all".to_string()),
+ ("heartbeat.interval".to_string(), "30s".to_string()),
("metadata.journal_slots".to_string(), "1024".to_string()),
],
tls: None,
@@ -762,7 +762,7 @@ mod tests {
};
assert_eq!(
v.suffix(),
- "http_segment_cache_indexes_all_metadata_journal_slots_1024"
+ "http_heartbeat_interval_30s_metadata_journal_slots_1024"
);
}
@@ -825,8 +825,8 @@ mod tests {
server: crate::attrs::ServerAttrs {
config_overrides: vec![
ConfigOverride {
- path: "segment.cache_indexes".to_string(),
- value: ConfigValue::Matrix(vec!["none".to_string(), "all".to_string()]),
+ path: "heartbeat.interval".to_string(),
+ value: ConfigValue::Matrix(vec!["30s".to_string(), "60s".to_string()]),
},
ConfigOverride {
path: "partition.validate_checksum".to_string(),
@@ -840,7 +840,7 @@ mod tests {
jwks_server: None,
};
let variants = generate_variants(&attrs);
- // 2 transports * 2 cache modes * 2 checksum modes = 8 variants
+ // 2 transports * 2 intervals * 2 checksum modes = 8 variants
assert_eq!(variants.len(), 8);
}
@@ -853,18 +853,18 @@ mod tests {
#[test]
fn cartesian_product_single() {
let overrides = vec![ConfigOverride {
- path: "segment.cache_indexes".to_string(),
- value: ConfigValue::Matrix(vec!["none".to_string(), "all".to_string()]),
+ path: "heartbeat.interval".to_string(),
+ value: ConfigValue::Matrix(vec!["30s".to_string(), "60s".to_string()]),
}];
let result = cartesian_product(&overrides);
assert_eq!(result.len(), 2);
assert_eq!(
result[0],
- vec![("segment.cache_indexes".to_string(), "none".to_string())]
+ vec![("heartbeat.interval".to_string(), "30s".to_string())]
);
assert_eq!(
result[1],
- vec![("segment.cache_indexes".to_string(), "all".to_string())]
+ vec![("heartbeat.interval".to_string(), "60s".to_string())]
);
}
diff --git a/core/harness_derive/src/lib.rs b/core/harness_derive/src/lib.rs
index 757e02127..e02981603 100644
--- a/core/harness_derive/src/lib.rs
+++ b/core/harness_derive/src/lib.rs
@@ -31,7 +31,7 @@
//!
//! Test with transport matrix:
//! ```ignore
-//! #[iggy_harness(transport = [Tcp, Http, Quic, WebSocket])]
+//! #[iggy_harness(test_client_transport = [Tcp, Http, Quic, WebSocket])]
//! async fn test_all_transports(client: &IggyClient) {
//! client.ping().await.unwrap();
//! }
@@ -40,11 +40,11 @@
//! Test with server config matrix:
//! ```ignore
//! #[iggy_harness(server(
-//! segment.cache_indexes = ["none", "all"],
+//! heartbeat.interval = ["30s", "60s"],
//! partition.validate_checksum = [true, false],
//! ))]
-//! async fn test_caching(client: &IggyClient) {
-//! // 2 cache modes × 2 checksum modes = 4 tests
+//! async fn test_persistence(client: &IggyClient) {
+//! // 2 intervals × 2 checksum modes = 4 tests
//! }
//! ```
@@ -65,7 +65,7 @@ use codegen::generate_tests;
///
/// # Attributes
///
-/// - `transport = [Tcp, Http, Quic, WebSocket]` - Generate variants for each transport
+/// - `test_client_transport = [Tcp, Http, Quic, WebSocket]` - Generate variants for each transport
/// - `server(key = value)` - Static server config
/// - `server(key = [v1, v2])` - Generate variants for each config value
///
diff --git a/core/integration/src/harness/config/resolve.rs b/core/integration/src/harness/config/resolve.rs
index 5ca0ccbc1..4f35b8535 100644
--- a/core/integration/src/harness/config/resolve.rs
+++ b/core/integration/src/harness/config/resolve.rs
@@ -40,8 +40,9 @@ pub const NON_CONFIG_ENV_VARS: [&str; 4] = [
///
/// # Implicit defaults
///
-/// - Setting any `tcp.socket.*` option automatically enables `tcp.socket.override_defaults`
-/// - Setting any `quic.socket.*` option automatically enables `quic.socket.override_defaults`
+/// - `encryption` is shorthand for `system.encryption.key`.
+/// - Setting that key also turns `system.encryption.enabled` on, unless the
+/// caller passed it explicitly.
///
/// # Errors
///
@@ -50,8 +51,6 @@ pub fn resolve_config_paths(
overrides: &HashMap<String, String>,
) -> Result<HashMap<String, String>, String> {
let mut env_vars = HashMap::new();
- let mut needs_tcp_socket_override = false;
- let mut needs_quic_socket_override = false;
let mut needs_encryption_enabled = false;
for (path, value) in overrides {
@@ -69,13 +68,6 @@ pub fn resolve_config_paths(
Some(m) => {
env_vars.insert(m.env_name.to_string(), value.clone());
- // Track if we need to enable socket override_defaults
- if path.starts_with("tcp.socket.") && path != "tcp.socket.override_defaults" {
- needs_tcp_socket_override = true;
- }
- if path.starts_with("quic.socket.") && path != "quic.socket.override_defaults" {
- needs_quic_socket_override = true;
- }
// Track if encryption key is set (auto-enable encryption)
if path == "encryption"
|| path == "encryption.key"
@@ -101,17 +93,6 @@ pub fn resolve_config_paths(
}
}
- // Auto-enable override_defaults when socket settings are customized
- if needs_tcp_socket_override && let Some(m) = find_mapping("tcp.socket.override_defaults") {
- env_vars
- .entry(m.env_name.to_string())
- .or_insert_with(|| "true".to_string());
- }
- if needs_quic_socket_override && let Some(m) = find_mapping("quic.socket.override_defaults") {
- env_vars
- .entry(m.env_name.to_string())
- .or_insert_with(|| "true".to_string());
- }
// Auto-enable encryption when key is set
if needs_encryption_enabled && let Some(m) = find_mapping("system.encryption.enabled") {
env_vars
diff --git a/core/integration/src/harness/handle/server.rs b/core/integration/src/harness/handle/server.rs
index bb6a6486d..98c1f014a 100644
--- a/core/integration/src/harness/handle/server.rs
+++ b/core/integration/src/harness/handle/server.rs
@@ -18,7 +18,7 @@
use super::client_builder::{ClientBuilder, ServerConnection};
use super::connectors_runtime::ConnectorsRuntimeHandle;
use super::mcp::McpHandle;
-use crate::harness::config::{ConnectorsRuntimeConfig, IpAddrKind, McpConfig, TestServerConfig};
+use crate::harness::config::{ConnectorsRuntimeConfig, McpConfig, TestServerConfig};
use crate::harness::context::TestContext;
use crate::harness::error::TestBinaryError;
use crate::harness::port_reserver::PortReserver;
@@ -297,12 +297,6 @@ impl ServerHandle {
.entry("IGGY_SYSTEM_SHARDING_CPU_ALLOCATION".to_string())
.or_insert(cpu_allocation);
- if self.config.ip_kind == IpAddrKind::V6 {
- self.envs
- .entry("IGGY_TCP_IPV6".to_string())
- .or_insert_with(|| "true".to_string());
- }
-
self.envs
.entry("IGGY_ROOT_USERNAME".to_string())
.or_insert_with(|| DEFAULT_ROOT_USERNAME.to_string());
diff --git a/core/integration/tests/config_provider/mod.rs b/core/integration/tests/config_provider/mod.rs
index a3bfcbe83..db7472d7a 100644
--- a/core/integration/tests/config_provider/mod.rs
+++ b/core/integration/tests/config_provider/mod.rs
@@ -31,16 +31,11 @@ use std::path::PathBuf;
async fn validate_config_env_override() {
let expected_http = true;
let expected_tcp = true;
- let expected_message_saver = true;
let expected_validate_checksum = false;
unsafe {
env::set_var("IGGY_HTTP_ENABLED", expected_http.to_string());
env::set_var("IGGY_TCP_ENABLED", expected_tcp.to_string());
- env::set_var(
- "IGGY_MESSAGE_SAVER_ENABLED",
- expected_message_saver.to_string(),
- );
env::set_var(
"IGGY_SYSTEM_PARTITION_VALIDATE_CHECKSUM",
expected_validate_checksum.to_string(),
@@ -57,7 +52,6 @@ async fn validate_config_env_override() {
assert_eq!(config.http.enabled, expected_http);
assert_eq!(config.tcp.enabled, expected_tcp);
- assert_eq!(config.message_saver.enabled, expected_message_saver);
assert_eq!(
config.system.partition.validate_checksum,
expected_validate_checksum
@@ -66,26 +60,22 @@ async fn validate_config_env_override() {
unsafe {
env::remove_var("IGGY_HTTP_ENABLED");
env::remove_var("IGGY_TCP_ENABLED");
- env::remove_var("IGGY_MESSAGE_SAVER_ENABLED");
env::remove_var("IGGY_SYSTEM_PARTITION_VALIDATE_CHECKSUM");
}
}
#[serial]
#[tokio::test]
-async fn validate_socket_override() {
- // Environment variables are set as raw byte counts
- let send_buffer_bytes = 666666_u64;
- let recv_buffer_bytes = 777777_u64;
+async fn validate_byte_size_leaf_override() {
+ // A non-primitive `#[config_env(leaf)]` field reads one env var rather than
+ // expanding into member variables. Dropping the attribute would not compile
+ // for `IggyByteSize`, so what needs pinning is the value format: the
+ // environment carries a raw byte count, not a formatted size string.
+ let stream_receive_window_bytes = 8_388_608_u64;
unsafe {
- env::set_var("IGGY_TCP_SOCKET_OVERRIDE_DEFAULTS", "true");
env::set_var(
- "IGGY_TCP_SOCKET_SEND_BUFFER_SIZE",
- send_buffer_bytes.to_string(),
- );
- env::set_var(
- "IGGY_TCP_SOCKET_RECV_BUFFER_SIZE",
- recv_buffer_bytes.to_string(),
+ "IGGY_QUIC_STREAM_RECEIVE_WINDOW",
+ stream_receive_window_bytes.to_string(),
);
}
@@ -95,40 +85,18 @@ async fn validate_socket_override() {
let config: ServerConfig = file_config_provider
.load_config()
.await
- .expect("Failed to load config.toml config with socket override");
+ .expect("Failed to load config.toml config with byte size leaf override");
- assert!(config.tcp.socket.override_defaults);
- // Verify the buffer sizes match the expected byte counts
assert_eq!(
- config.tcp.socket.send_buffer_size.as_bytes_u64(),
- send_buffer_bytes
- );
- assert_eq!(
- config.tcp.socket.recv_buffer_size.as_bytes_u64(),
- recv_buffer_bytes
+ config.quic.stream_receive_window.as_bytes_u64(),
+ stream_receive_window_bytes
);
unsafe {
- env::remove_var("IGGY_TCP_SOCKET_OVERRIDE_DEFAULTS");
- env::remove_var("IGGY_TCP_SOCKET_SEND_BUFFER_SIZE");
- env::remove_var("IGGY_TCP_SOCKET_RECV_BUFFER_SIZE");
+ env::remove_var("IGGY_QUIC_STREAM_RECEIVE_WINDOW");
}
}
-#[serial]
-#[tokio::test]
-async fn validate_socket_no_override() {
- let config_path = get_root_path().join("../server/config.toml");
- let file_config_provider =
- ServerConfig::config_provider(&config_path.as_path().display().to_string());
- let config: ServerConfig = file_config_provider
- .load_config()
- .await
- .expect("Failed to load config.toml config without socket override");
-
- assert!(!config.tcp.socket.override_defaults);
-}
-
#[serial]
#[tokio::test]
async fn validate_cluster_config_env_override() {
diff --git a/core/integration/tests/data_integrity/verify_after_server_restart.rs b/core/integration/tests/data_integrity/verify_after_server_restart.rs
index 1ac9b91fa..e831eae1f 100644
--- a/core/integration/tests/data_integrity/verify_after_server_restart.rs
+++ b/core/integration/tests/data_integrity/verify_after_server_restart.rs
@@ -19,29 +19,7 @@ use iggy::prelude::*;
use integration::bench_utils::run_bench_and_wait_for_finish;
use integration::harness::{TestHarness, TestServerConfig};
use serial_test::parallel;
-use std::{collections::HashMap, str::FromStr};
-use test_case::test_matrix;
-
-fn cache_open_segment() -> &'static str {
- "open_segment"
-}
-
-fn cache_all() -> &'static str {
- "all"
-}
-
-fn cache_none() -> &'static str {
- "none"
-}
-
-fn build_server_config(cache_setting: &str) -> TestServerConfig {
- let mut extra_envs = HashMap::new();
- extra_envs.insert(
- "IGGY_SYSTEM_SEGMENT_CACHE_INDEXES".to_string(),
- cache_setting.to_string(),
- );
- TestServerConfig::builder().extra_envs(extra_envs).build()
-}
+use std::str::FromStr;
// TODO(numminex) - Move the message generation method from benchmark run to a special method.
//
@@ -53,18 +31,15 @@ fn build_server_config(cache_setting: &str) -> TestServerConfig {
// `iggy-bench` must be freshly built: the harness spawns the prebuilt binary,
// and a stale one never completes a frame against the server, tripping the
// bench timeout in `run_bench_and_wait_for_finish`.
-#[test_matrix(
- [cache_all(), cache_open_segment(), cache_none()]
-)]
#[tokio::test]
#[parallel]
-async fn should_fill_data_and_verify_after_restart(cache_setting: &'static str) {
+async fn should_fill_data_and_verify_after_restart() {
// Restart scenarios run single-node: restarting a node in a multi-node
// cluster trips a known partitions-plane view-change stall, tracked
// separately.
let mut harness = TestHarness::builder()
.cluster_nodes(1)
- .server(build_server_config(cache_setting))
+ .server(TestServerConfig::default())
.build()
.unwrap();
diff --git a/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs b/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs
index f4f935a82..d392755ae 100644
--- a/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs
+++ b/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs
@@ -86,8 +86,6 @@ async fn create_tcp_client(server_addr: &str) -> IggyClient {
// Deliberately short: this spec drives the server's eviction path (see the
// module note), so the verifier must be able to reap a stale member.
heartbeat.interval = "2s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_not_duplicate_partition_assignments_after_stale_client_cleanup(
harness: &TestHarness,
@@ -275,8 +273,6 @@ async fn should_not_duplicate_partition_assignments_after_stale_client_cleanup(
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_not_reshuffle_partitions_when_new_member_joins(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -353,8 +349,6 @@ async fn should_not_reshuffle_partitions_when_new_member_joins(harness: &TestHar
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_skip_revoked_partitions_in_round_robin(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -435,8 +429,6 @@ async fn should_skip_revoked_partitions_in_round_robin(harness: &TestHarness) {
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_not_lose_messages_with_concurrent_polls_during_partition_add(
harness: &TestHarness,
@@ -595,8 +587,6 @@ async fn should_not_lose_messages_with_concurrent_polls_during_partition_add(
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_handle_partition_add_then_consumer_disconnect_then_new_join(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -743,8 +733,6 @@ async fn should_handle_partition_add_then_consumer_disconnect_then_new_join(harn
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_handle_partition_delete_while_multiple_consumers_polling(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -880,8 +868,6 @@ async fn should_handle_partition_delete_while_multiple_consumers_polling(harness
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_reach_even_distribution_after_multiple_joins(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -1062,8 +1048,6 @@ async fn should_reach_even_distribution_after_multiple_joins(harness: &TestHarne
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_split_evenly_when_consumer_joins_after_partitions_added(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -1183,8 +1167,6 @@ async fn should_split_evenly_when_consumer_joins_after_partitions_added(harness:
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_not_duplicate_messages_when_partitions_added_during_polling(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -1284,8 +1266,6 @@ async fn should_not_duplicate_messages_when_partitions_added_during_polling(harn
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_handle_delete_partitions_with_uncommitted_work(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -1416,8 +1396,6 @@ async fn should_handle_delete_partitions_with_uncommitted_work(harness: &TestHar
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_handle_rapid_partition_changes_with_active_consumers(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -1554,8 +1532,6 @@ async fn should_handle_rapid_partition_changes_with_active_consumers(harness: &T
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_rebalance_after_adding_partitions(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -1628,8 +1604,6 @@ async fn should_rebalance_after_adding_partitions(harness: &TestHarness) {
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_rebalance_after_deleting_partitions(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -1716,8 +1690,6 @@ async fn should_rebalance_after_deleting_partitions(harness: &TestHarness) {
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_handle_partition_add_during_pending_revocation(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -1798,8 +1770,7 @@ async fn should_handle_partition_add_during_pending_revocation(harness: &TestHar
}
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
- consumer_group.rebalancing_timeout = "3s",
- consumer_group.rebalancing_check_interval = "1s"
+ consumer_group.rebalancing_timeout = "3s"
))]
async fn should_timeout_revocation(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -1909,8 +1880,6 @@ async fn should_timeout_revocation(harness: &TestHarness) {
// Deliberately short: this spec drives the server's eviction path (see the
// module note), so the verifier must be able to reap a stale member.
heartbeat.interval = "2s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_not_duplicate_after_reconnect_without_heartbeat(harness: &TestHarness) {
let server_addr = harness.server().raw_tcp_addr().expect("tcp addr");
@@ -2052,10 +2021,7 @@ async fn should_not_duplicate_after_reconnect_without_heartbeat(harness: &TestHa
.unwrap();
}
-#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
-))]
+#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic])]
async fn should_not_duplicate_partition_assignments_after_client_reconnect(harness: &TestHarness) {
let root_client = harness
.root_client()
@@ -2353,8 +2319,6 @@ fn assert_balanced_partition_distribution(cg: &ConsumerGroupDetails, expected_to
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_not_return_same_message_to_two_consumers_during_rebalance(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -2462,8 +2426,6 @@ async fn should_not_return_same_message_to_two_consumers_during_rebalance(harnes
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_complete_revocation_on_auto_commit(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -2534,8 +2496,6 @@ async fn should_complete_revocation_on_auto_commit(harness: &TestHarness) {
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_transfer_never_polled_partitions_immediately(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -2592,8 +2552,6 @@ async fn should_transfer_never_polled_partitions_immediately(harness: &TestHarne
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_rebalance_when_member_with_pending_revocation_leaves(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -2684,8 +2642,6 @@ async fn should_rebalance_when_member_with_pending_revocation_leaves(harness: &T
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_not_produce_duplicate_messages_with_sequential_consumer_joins(
harness: &TestHarness,
@@ -2842,8 +2798,6 @@ async fn should_not_produce_duplicate_messages_with_sequential_consumer_joins(
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_wait_for_manual_commit_before_completing_revocation(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -2951,8 +2905,6 @@ async fn should_wait_for_manual_commit_before_completing_revocation(harness: &Te
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_redistribute_when_revocation_target_leaves(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -3044,8 +2996,6 @@ async fn should_redistribute_when_revocation_target_leaves(harness: &TestHarness
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_distribute_partitions_evenly_with_concurrent_joins(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -3144,8 +3094,6 @@ async fn should_distribute_partitions_evenly_with_concurrent_joins(harness: &Tes
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_not_assign_partition_to_wrong_member_after_slab_reuse(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -3250,8 +3198,6 @@ async fn should_not_assign_partition_to_wrong_member_after_slab_reuse(harness: &
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_not_complete_other_members_revocations_on_leave(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -3381,8 +3327,6 @@ async fn should_not_complete_other_members_revocations_on_leave(harness: &TestHa
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_distribute_16_partitions_evenly_across_16_consumers(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -3474,8 +3418,6 @@ async fn should_distribute_16_partitions_evenly_across_16_consumers(harness: &Te
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_distribute_excess_evenly_when_multiple_idle_members_join(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -3541,8 +3483,6 @@ async fn should_distribute_excess_evenly_when_multiple_idle_members_join(harness
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_distribute_remainder_fairly_with_uneven_ratio(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -3606,8 +3546,6 @@ async fn should_distribute_remainder_fairly_with_uneven_ratio(harness: &TestHarn
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_collect_excess_from_multiple_overassigned_members(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -3667,8 +3605,6 @@ async fn should_collect_excess_from_multiple_overassigned_members(harness: &Test
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_not_starve_any_member_in_large_scale_rebalance(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
@@ -3722,8 +3658,6 @@ async fn should_not_starve_any_member_in_large_scale_rebalance(harness: &TestHar
#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic], server(
heartbeat.enabled = true,
heartbeat.interval = "60s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_maintain_balance_after_member_churn(harness: &TestHarness) {
let root_client = harness.root_client().await.unwrap();
diff --git a/core/integration/tests/sdk/consumer_group.rs b/core/integration/tests/sdk/consumer_group.rs
index c6aec3104..b670de3ee 100644
--- a/core/integration/tests/sdk/consumer_group.rs
+++ b/core/integration/tests/sdk/consumer_group.rs
@@ -30,7 +30,15 @@ const CONSUMER_USERNAME: &str = "consumer-group-rejoin-user";
const CONSUMER_PASSWORD: &str = "password123";
const CONSUMER_REJOIN_TIMEOUT: Duration = Duration::from_secs(10);
-#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic])]
+// Pins a 60s server heartbeat because harness clients never ping on their own:
+// the SDK pinger is spawned by `IggyClient::connect`, which the harness builder
+// does not call. At the shipped 30s interval an idle group member is reaped by
+// the server's verifier, and the failure surfaces as a short member count
+// instead of anything about the scenario under test.
+#[iggy_harness(
+ test_client_transport = [Tcp, WebSocket, Quic],
+ server(heartbeat.enabled = true, heartbeat.interval = "60s")
+)]
async fn consumer_group_retries_rejoin_after_failure(harness: &TestHarness) {
let root_client = harness
.root_client()
diff --git a/core/integration/tests/sdk/consumer_group_membership.rs b/core/integration/tests/sdk/consumer_group_membership.rs
index 05342d3b1..429aea649 100644
--- a/core/integration/tests/sdk/consumer_group_membership.rs
+++ b/core/integration/tests/sdk/consumer_group_membership.rs
@@ -40,12 +40,22 @@ const TOPIC_ID_NOT_FOUND: u32 = 2010;
const CONSUMER_GROUP_ID_NOT_FOUND: u32 = 5000;
const CONSUMER_GROUP_MEMBER_NOT_FOUND: u32 = 5006;
+// Every spec here pins a 60s server heartbeat because harness clients never
+// ping on their own: the SDK pinger is spawned by `IggyClient::connect`, which
+// the harness builder does not call. At the shipped 30s interval a group member
+// that idles through another client's setup is reaped by the server's verifier,
+// and the failure surfaces as a short member count instead of anything about
+// the scenario under test.
+
// A consumer-group member holding zero partitions has the same empty client-side
// assignment as a non-member; only membership tells them apart. When the group
// is deleted under such a member, the poll must surface an error (driving a
// rejoin) rather than treating the empty assignment as "nothing to poll" and
// hanging forever.
-#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic])]
+#[iggy_harness(
+ test_client_transport = [Tcp, WebSocket, Quic],
+ server(heartbeat.enabled = true, heartbeat.interval = "60s")
+)]
async fn given_group_member_holds_no_partitions_when_group_deleted_should_surface_error_not_hang(
harness: &TestHarness,
) {
@@ -152,7 +162,10 @@ async fn given_group_member_holds_no_partitions_when_group_deleted_should_surfac
// codes the legacy server returns. Binary transports only: the HTTP client
// has no join/leave (stateless sessions carry no member identity, the SDK
// returns FeatureUnavailable client-side), so the ladder cannot run there.
-#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic])]
+#[iggy_harness(
+ test_client_transport = [Tcp, WebSocket, Quic],
+ server(heartbeat.enabled = true, heartbeat.interval = "60s")
+)]
async fn given_join_and_leave_failures_when_sent_over_the_wire_should_return_legacy_error_codes(
harness: &TestHarness,
) {
diff --git a/core/integration/tests/server/cg.rs b/core/integration/tests/server/cg.rs
index 697b3ddd9..eb5382e59 100644
--- a/core/integration/tests/server/cg.rs
+++ b/core/integration/tests/server/cg.rs
@@ -26,9 +26,16 @@ use integration::iggy_harness;
// Consumer group scenarios do not support HTTP (stateful operations).
+// Every spec here pins a 60s server heartbeat because harness clients never
+// ping on their own: the SDK pinger is spawned by `IggyClient::connect`, which
+// the harness builder does not call. At the shipped 30s interval a group member
+// that idles through another client's setup is reaped by the server's verifier,
+// and the failure surfaces as a short member count instead of anything about
+// the scenario under test.
+
#[iggy_harness(
test_client_transport = [Tcp, WebSocket, Quic],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ server(heartbeat.enabled = true, heartbeat.interval = "60s")
)]
async fn join(harness: &TestHarness) {
consumer_group_join_scenario::run(harness).await;
@@ -36,7 +43,7 @@ async fn join(harness: &TestHarness) {
#[iggy_harness(
test_client_transport = [Tcp, WebSocket, Quic],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ server(heartbeat.enabled = true, heartbeat.interval = "60s")
)]
async fn single_client(harness: &TestHarness) {
consumer_group_with_single_client_polling_messages_scenario::run(harness).await;
@@ -44,7 +51,7 @@ async fn single_client(harness: &TestHarness) {
#[iggy_harness(
test_client_transport = [Tcp, WebSocket, Quic],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ server(heartbeat.enabled = true, heartbeat.interval = "60s")
)]
async fn multiple_clients(harness: &TestHarness) {
consumer_group_with_multiple_clients_polling_messages_scenario::run(harness).await;
@@ -52,7 +59,7 @@ async fn multiple_clients(harness: &TestHarness) {
#[iggy_harness(
test_client_transport = [Tcp, WebSocket, Quic],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ server(heartbeat.enabled = true, heartbeat.interval = "60s")
)]
async fn auto_commit_reconnection(harness: &TestHarness) {
consumer_group_auto_commit_reconnection_scenario::run(harness).await;
@@ -60,7 +67,7 @@ async fn auto_commit_reconnection(harness: &TestHarness) {
#[iggy_harness(
test_client_transport = [Tcp, WebSocket, Quic],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ server(heartbeat.enabled = true, heartbeat.interval = "60s")
)]
async fn new_messages_after_restart(harness: &TestHarness) {
consumer_group_new_messages_after_restart_scenario::run(harness).await;
@@ -68,7 +75,7 @@ async fn new_messages_after_restart(harness: &TestHarness) {
#[iggy_harness(
test_client_transport = [Tcp, WebSocket, Quic],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ server(heartbeat.enabled = true, heartbeat.interval = "60s")
)]
async fn offset_cleanup(harness: &TestHarness) {
consumer_group_offset_cleanup_scenario::run(harness).await;
@@ -76,7 +83,7 @@ async fn offset_cleanup(harness: &TestHarness) {
#[iggy_harness(
test_client_transport = [Tcp, WebSocket, Quic],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ server(heartbeat.enabled = true, heartbeat.interval = "60s")
)]
async fn duplicate_name_create_preserves_live_group(harness: &TestHarness) {
consumer_group_duplicate_name_create_scenario::run(harness).await;
diff --git a/core/integration/tests/server/flush_vsr.rs b/core/integration/tests/server/flush_vsr.rs
index a40e0b5e1..adeee0dd7 100644
--- a/core/integration/tests/server/flush_vsr.rs
+++ b/core/integration/tests/server/flush_vsr.rs
@@ -31,8 +31,7 @@ use integration::iggy_harness;
const PARTITION_ID: u32 = 0;
#[iggy_harness(
- test_client_transport = [Tcp],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ test_client_transport = [Tcp]
)]
async fn given_valid_partition_when_flushing_should_reject_feature_unavailable(
harness: &TestHarness,
diff --git a/core/integration/tests/server/general.rs b/core/integration/tests/server/general.rs
index 460c0efdb..861d24895 100644
--- a/core/integration/tests/server/general.rs
+++ b/core/integration/tests/server/general.rs
@@ -25,8 +25,6 @@ use integration::iggy_harness;
#[iggy_harness(
test_client_transport = [Tcp, Http, Quic, WebSocket],
server(
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true,
quic.max_idle_timeout = "500s",
quic.keep_alive_interval = "15s"
)
@@ -38,8 +36,6 @@ async fn authentication(harness: &TestHarness) {
#[iggy_harness(
test_client_transport = [Tcp, Http, Quic, WebSocket],
server(
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true,
quic.max_idle_timeout = "500s",
quic.keep_alive_interval = "15s"
)
@@ -51,8 +47,6 @@ async fn system(harness: &TestHarness) {
#[iggy_harness(
test_client_transport = [Tcp, Http, Quic, WebSocket],
server(
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true,
quic.max_idle_timeout = "500s",
quic.keep_alive_interval = "15s"
)
@@ -64,8 +58,6 @@ async fn user(harness: &TestHarness) {
#[iggy_harness(
test_client_transport = [Tcp, Http, Quic, WebSocket],
server(
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true,
quic.max_idle_timeout = "500s",
quic.keep_alive_interval = "15s"
)
@@ -77,8 +69,6 @@ async fn permissions(harness: &TestHarness) {
#[iggy_harness(
test_client_transport = [Tcp, Http, Quic, WebSocket],
server(
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true,
quic.max_idle_timeout = "500s",
quic.keep_alive_interval = "15s"
)
@@ -90,8 +80,6 @@ async fn message_headers(harness: &TestHarness) {
#[iggy_harness(
test_client_transport = [Tcp, Http, Quic, WebSocket],
server(
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true,
quic.max_idle_timeout = "500s",
quic.keep_alive_interval = "15s"
)
@@ -103,8 +91,6 @@ async fn stream_size_validation(harness: &TestHarness) {
#[iggy_harness(
test_client_transport = [Tcp, Http, Quic, WebSocket],
server(
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true,
quic.max_idle_timeout = "500s",
quic.keep_alive_interval = "15s"
)
@@ -116,8 +102,6 @@ async fn consumer_timestamp_polling(harness: &TestHarness) {
#[iggy_harness(
test_client_transport = [Tcp, Http, Quic, WebSocket],
server(
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true,
quic.max_idle_timeout = "500s",
quic.keep_alive_interval = "15s"
)
@@ -133,8 +117,6 @@ async fn snapshot(harness: &TestHarness) {
#[iggy_harness(
test_client_transport = [Tcp, Http, Quic, WebSocket],
server(
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true,
quic.max_idle_timeout = "500s",
quic.keep_alive_interval = "15s"
)
diff --git a/core/integration/tests/server/http_client.rs b/core/integration/tests/server/http_client.rs
index 7393ffef3..2ba24ac32 100644
--- a/core/integration/tests/server/http_client.rs
+++ b/core/integration/tests/server/http_client.rs
@@ -156,7 +156,8 @@ impl HttpClient {
.expect("get request")
}
- /// Unauthenticated GET (no bearer) for the public `/ping` probe.
+ /// Unauthenticated GET (no bearer): the public `/ping` probe and 401
+ /// assertions on protected routes.
pub async fn get_anonymous(&self, path: &str) -> Response {
self.client
.get(self.url(path))
diff --git a/core/integration/tests/server/http_rbac.rs b/core/integration/tests/server/http_rbac.rs
index 6b949905e..1a5728ba3 100644
--- a/core/integration/tests/server/http_rbac.rs
+++ b/core/integration/tests/server/http_rbac.rs
@@ -25,7 +25,8 @@
//! - the SDK-hidden success codes (produce 201, reads 200, mutations 204, and
//! the server-assigned id in a 200 create-stream body);
//! - HTTP-only route shapes: the `/users/me` self alias, the caller-scoped
-//! name-sorted PAT-list array, auth-only cluster metadata, and the public ping;
+//! name-sorted PAT-list array, auth-only cluster metadata and metrics scrape,
+//! and the public ping;
//! - the change-password status/body shape: a wrong current password renders a
//! typed 400 `InvalidCredentials` and a missing target a 404 `ResourceNotFound`,
//! both bodies the SDK hides. Change-password commits over every transport, so
@@ -405,6 +406,25 @@ async fn given_http_only_routes_when_requested_should_return_expected_shapes(
"ping is public"
);
+ // The metrics scrape is auth-only like cluster metadata: anonymous is 401,
+ // an ungranted caller reads the exposition.
+ assert_eq!(
+ alice.get_anonymous("/metrics").await.status(),
+ StatusCode::UNAUTHORIZED,
+ "metrics scrape without a bearer must be 401"
+ );
+ let metrics = alice.get("/metrics").await;
+ assert_eq!(
+ metrics.status(),
+ StatusCode::OK,
+ "metrics scrape is auth-only, never RBAC-gated"
+ );
+ let exposition = metrics.text().await.expect("metrics exposition body");
+ assert!(
+ exposition.contains("# TYPE streams "),
+ "metrics exposition must carry the parity metric set:\n{exposition}"
+ );
+
// PAT listing is a JSON array of the caller's own tokens, name-sorted
// (created in reverse order to prove the sort). Create returns 200.
for name in ["alice-zeta", "alice-alpha"] {
diff --git a/core/integration/tests/server/message_cleanup.rs b/core/integration/tests/server/message_cleanup.rs
index fad8ab490..9677b7dda 100644
--- a/core/integration/tests/server/message_cleanup.rs
+++ b/core/integration/tests/server/message_cleanup.rs
@@ -93,6 +93,18 @@ async fn run_cleanup_scenario(scenario: CleanupScenarioFn) {
"IGGY_DATA_MAINTENANCE_MESSAGES_INTERVAL".to_string(),
"100ms".to_string(),
),
+ // Size retention floors each partition's budget at one
+ // SEALED segment, `segment_size + max_message_size`, so the
+ // shipped 64 MiB bus cap would put the floor far above the
+ // few MiB these scenarios produce and no size-based
+ // deletion could ever fire. 2 MiB keeps the floor under the
+ // topic caps below while still clearing the largest request
+ // any scenario sends (a 10-message batch, ~1.002 MiB).
+ // Byte-size env leaves carry a raw byte count.
+ (
+ "IGGY_MESSAGE_BUS_MAX_MESSAGE_SIZE".to_string(),
+ (2 * 1024 * 1024).to_string(),
+ ),
]))
.build(),
)
diff --git a/core/integration/tests/server/message_retrieval.rs b/core/integration/tests/server/message_retrieval.rs
index 2546dad97..8b1d85978 100644
--- a/core/integration/tests/server/message_retrieval.rs
+++ b/core/integration/tests/server/message_retrieval.rs
@@ -19,7 +19,6 @@ use crate::server::scenarios::{offset_scenario, timestamp_scenario};
use iggy::prelude::*;
use integration::harness::{TestHarness, TestServerConfig};
use serial_test::parallel;
-use std::collections::HashMap;
use test_case::test_matrix;
// A topic's segment size must be a 512-byte multiple in
@@ -39,18 +38,6 @@ fn segment_size_10mib() -> u64 {
10 * 1024 * 1024
}
-fn cache_none() -> &'static str {
- "none"
-}
-
-fn cache_all() -> &'static str {
- "all"
-}
-
-fn cache_open_segment() -> &'static str {
- "open_segment"
-}
-
fn msgs_req_32() -> u32 {
32
}
@@ -81,46 +68,26 @@ fn topic_options(segment_size: u64, messages_required_to_save: u32) -> TopicCrea
}
}
-fn build_server_config(cache_indexes: &str) -> TestServerConfig {
- let mut extra_envs = HashMap::new();
- extra_envs.insert(
- "IGGY_SYSTEM_SEGMENT_CACHE_INDEXES".to_string(),
- cache_indexes.to_string(),
- );
- extra_envs.insert(
- "IGGY_TCP_SOCKET_OVERRIDE_DEFAULTS".to_string(),
- "true".to_string(),
- );
- extra_envs.insert("IGGY_TCP_SOCKET_NODELAY".to_string(), "true".to_string());
-
- TestServerConfig::builder().extra_envs(extra_envs).build()
-}
-
/// These matrices exercise retrieval, not replication: a single node keeps
/// the wide permutation set cheap under a parallel nextest run, where an
/// oversubscribed multi-node cluster stalls past the liveness window and
/// elects a new primary mid-scenario, killing the client session.
-fn build_harness(cache_indexes: &str) -> TestHarness {
+fn build_harness() -> TestHarness {
TestHarness::builder()
- .server(build_server_config(cache_indexes))
.cluster_nodes(1)
+ .server(TestServerConfig::default())
.build()
.unwrap()
}
#[test_matrix(
[segment_size_1mib(), segment_size_2mib(), segment_size_10mib()],
- [cache_none(), cache_all(), cache_open_segment()],
[msgs_req_32(), msgs_req_64(), msgs_req_1024(), msgs_req_9984()]
)]
#[tokio::test]
#[parallel]
-async fn get_by_offset_scenario(
- segment_size: u64,
- cache_indexes: &str,
- messages_required_to_save: u32,
-) {
- let mut harness = build_harness(cache_indexes);
+async fn get_by_offset_scenario(segment_size: u64, messages_required_to_save: u32) {
+ let mut harness = build_harness();
harness.start().await.unwrap();
@@ -133,17 +100,12 @@ async fn get_by_offset_scenario(
#[test_matrix(
[segment_size_1mib(), segment_size_2mib(), segment_size_10mib()],
- [cache_none(), cache_all(), cache_open_segment()],
[msgs_req_32(), msgs_req_64(), msgs_req_1024(), msgs_req_9984()]
)]
#[tokio::test]
#[parallel]
-async fn get_by_timestamp_scenario(
- segment_size: u64,
- cache_indexes: &str,
- messages_required_to_save: u32,
-) {
- let mut harness = build_harness(cache_indexes);
+async fn get_by_timestamp_scenario(segment_size: u64, messages_required_to_save: u32) {
+ let mut harness = build_harness();
harness.start().await.unwrap();
diff --git a/core/integration/tests/server/poll_semantics_vsr.rs b/core/integration/tests/server/poll_semantics_vsr.rs
index a2eb5e724..a3a90f06d 100644
--- a/core/integration/tests/server/poll_semantics_vsr.rs
+++ b/core/integration/tests/server/poll_semantics_vsr.rs
@@ -29,8 +29,7 @@ use integration::iggy_harness;
use tokio::time::{Duration, sleep};
#[iggy_harness(
- test_client_transport = [Tcp],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ test_client_transport = [Tcp]
)]
async fn given_missing_partition_when_polling_should_reject_partition_not_found(
harness: &TestHarness,
@@ -90,8 +89,7 @@ async fn given_missing_partition_when_polling_should_reject_partition_not_found(
}
#[iggy_harness(
- test_client_transport = [Tcp],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ test_client_transport = [Tcp]
)]
async fn given_missing_stream_when_polling_should_reject_stream_not_found(harness: &TestHarness) {
let client = harness.tcp_root_client().await.expect("tcp root client");
@@ -118,8 +116,7 @@ async fn given_missing_stream_when_polling_should_reject_stream_not_found(harnes
}
#[iggy_harness(
- test_client_transport = [Tcp],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ test_client_transport = [Tcp]
)]
async fn given_missing_topic_when_polling_should_reject_topic_not_found(harness: &TestHarness) {
let client = harness.tcp_root_client().await.expect("tcp root client");
@@ -157,8 +154,7 @@ async fn given_missing_topic_when_polling_should_reject_topic_not_found(harness:
/// fresh consumer. Legacy swallows this one too; the server surfaces the code
/// the poll path already surfaces for the identical addressing error.
#[iggy_harness(
- test_client_transport = [Tcp],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ test_client_transport = [Tcp]
)]
async fn given_missing_partition_when_getting_consumer_offset_should_reject_partition_not_found(
harness: &TestHarness,
@@ -208,8 +204,7 @@ async fn given_missing_partition_when_getting_consumer_offset_should_reject_part
}
#[iggy_harness(
- test_client_transport = [Tcp],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ test_client_transport = [Tcp]
)]
async fn given_message_at_polled_timestamp_when_polling_should_include_it(harness: &TestHarness) {
let client = harness.tcp_root_client().await.expect("tcp root client");
diff --git a/core/integration/tests/server/purge_delete.rs b/core/integration/tests/server/purge_delete.rs
index 0d503ce09..3b1e5ff08 100644
--- a/core/integration/tests/server/purge_delete.rs
+++ b/core/integration/tests/server/purge_delete.rs
@@ -21,12 +21,7 @@ use test_case::test_matrix;
// Restart scenarios run single-node: restarting a node in a multi-node cluster
// trips a known partitions-plane view-change stall, tracked separately.
-#[iggy_harness(
- cluster_nodes = 1,
- server(
- segment.cache_indexes = ["all", "none", "open_segment"],
- )
-)]
+#[iggy_harness(cluster_nodes = 1)]
#[test_matrix([restart_off(), restart_on()])]
async fn should_delete_segments_and_validate_filesystem(
harness: &mut TestHarness,
@@ -35,20 +30,13 @@ async fn should_delete_segments_and_validate_filesystem(
purge_delete_scenario::run(harness, restart_server).await;
}
-#[iggy_harness(
- cluster_nodes = 1,
- server(
- segment.cache_indexes = ["all", "none", "open_segment"],
- )
-)]
+#[iggy_harness(cluster_nodes = 1)]
#[test_matrix([restart_off(), restart_on()])]
async fn should_delete_segments_without_consumers(harness: &mut TestHarness, restart_server: bool) {
purge_delete_scenario::run_no_consumers(harness, restart_server).await;
}
-#[iggy_harness(server(
- segment.cache_indexes = ["all", "none", "open_segment"],
-))]
+#[iggy_harness]
async fn should_delete_segments_with_consumer_group_barrier(harness: &TestHarness) {
let client = harness.tcp_root_client().await.unwrap();
let data_path = harness.server().data_path();
@@ -56,12 +44,7 @@ async fn should_delete_segments_with_consumer_group_barrier(harness: &TestHarnes
purge_delete_scenario::run_consumer_group_barrier(&client, &data_path).await;
}
-#[iggy_harness(
- cluster_nodes = 1,
- server(
- segment.cache_indexes = ["all", "none", "open_segment"],
- )
-)]
+#[iggy_harness(cluster_nodes = 1)]
#[test_matrix([restart_off(), restart_on()])]
async fn should_block_deletion_until_all_consumers_pass_segment(
harness: &mut TestHarness,
@@ -70,12 +53,7 @@ async fn should_block_deletion_until_all_consumers_pass_segment(
purge_delete_scenario::run_multi_consumer_barrier(harness, restart_server).await;
}
-#[iggy_harness(
- cluster_nodes = 1,
- server(
- segment.cache_indexes = ["all", "none", "open_segment"],
- )
-)]
+#[iggy_harness(cluster_nodes = 1)]
// The scenario asserts the exact [0, 5, 10, 15, 20, 25] layout only on the legacy path;
// under vsr it verifies the framing-agnostic purge outcome (offsets cleared,
// files deleted, partition reset to a single segment at offset 0).
diff --git a/core/integration/tests/server/purge_vsr.rs b/core/integration/tests/server/purge_vsr.rs
index 152bd63bf..53df5b9e5 100644
--- a/core/integration/tests/server/purge_vsr.rs
+++ b/core/integration/tests/server/purge_vsr.rs
@@ -51,8 +51,7 @@ async fn given_journal_resident_messages_when_purged_should_not_resurface(
// read has to happen in the replicated apply. A retry loop here would pass
// against the pre-apply behavior too.
#[iggy_harness(
- test_client_transport = [Tcp],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ test_client_transport = [Tcp]
)]
async fn given_purged_topic_when_getting_topic_immediately_should_report_zero_stats(
harness: &TestHarness,
diff --git a/core/integration/tests/server/scenarios/concurrent_produce_consume_scenario.rs b/core/integration/tests/server/scenarios/concurrent_produce_consume_scenario.rs
index bbf8521cf..749399bb6 100644
--- a/core/integration/tests/server/scenarios/concurrent_produce_consume_scenario.rs
+++ b/core/integration/tests/server/scenarios/concurrent_produce_consume_scenario.rs
@@ -49,7 +49,7 @@ const MAX_TEST_DURATION: Duration = Duration::from_secs(60);
/// `messages_required_to_save = "1"` forces every batch to trigger an inline
/// journal commit. The next send arrives before async persist completes, creating
/// State C (journal non-empty + in-flight non-empty simultaneously).
-#[iggy_harness(server(message_saver.enabled = true, message_saver.interval = "100ms"))]
+#[iggy_harness]
async fn concurrent_produce_consume_no_offset_skip(harness: &TestHarness) {
let stream_id = Identifier::named(STREAM_NAME).unwrap();
let topic_id = Identifier::named(TOPIC_NAME).unwrap();
diff --git a/core/integration/tests/server/scenarios/cross_protocol_pat_scenario.rs b/core/integration/tests/server/scenarios/cross_protocol_pat_scenario.rs
index 1ef441cba..2709ce17d 100644
--- a/core/integration/tests/server/scenarios/cross_protocol_pat_scenario.rs
+++ b/core/integration/tests/server/scenarios/cross_protocol_pat_scenario.rs
@@ -35,10 +35,7 @@ async fn create_root_client(harness: &TestHarness, transport: TransportProtocol)
}
/// Create PAT via HTTP, list via multiple TCP connections.
-#[iggy_harness(server(
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
-))]
+#[iggy_harness]
pub async fn should_see_pat_created_via_http_when_listing_via_tcp(harness: &TestHarness) {
let http_client = create_root_client(harness, TransportProtocol::Http).await;
@@ -94,10 +91,7 @@ pub async fn should_see_pat_created_via_http_when_listing_via_tcp(harness: &Test
}
/// Create PAT via TCP, list via HTTP. Should work since TCP handlers broadcast.
-#[iggy_harness(server(
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
-))]
+#[iggy_harness]
pub async fn should_see_pat_created_via_tcp_when_listing_via_http(harness: &TestHarness) {
let tcp_client = create_root_client(harness, TransportProtocol::Tcp).await;
@@ -122,10 +116,7 @@ pub async fn should_see_pat_created_via_tcp_when_listing_via_http(harness: &Test
}
/// Create via TCP, delete via HTTP, verify deletion visible via TCP.
-#[iggy_harness(server(
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
-))]
+#[iggy_harness]
pub async fn should_not_see_pat_deleted_via_http_when_listing_via_tcp(harness: &TestHarness) {
let tcp_client = create_root_client(harness, TransportProtocol::Tcp).await;
diff --git a/core/integration/tests/server/scenarios/message_cleanup_scenario.rs b/core/integration/tests/server/scenarios/message_cleanup_scenario.rs
index 112f484cc..6c327a8bc 100644
--- a/core/integration/tests/server/scenarios/message_cleanup_scenario.rs
+++ b/core/integration/tests/server/scenarios/message_cleanup_scenario.rs
@@ -551,16 +551,20 @@ pub async fn run_expiry_with_multiple_partitions(client: &IggyClient, data_path:
.unwrap();
}
-/// Tests fair size-based cleanup across multiple partitions.
+/// Tests fair size-based cleanup across multiple partitions: the topic cap is
+/// enforced as a per-partition SHARE, not as a topic-wide total.
pub async fn run_fair_size_based_cleanup_multipartition(client: &IggyClient, data_path: &Path) {
const PARTITIONS_COUNT: u32 = 3;
let stream = client.create_stream(STREAM_NAME).await.unwrap();
let stream_id = stream.id;
- // 6 MiB max, cleanup at 90% = 5.4 MiB. Above `PARTITIONS_COUNT` * the 1 MiB
- // segment size, so every partition can seal a segment before the cap trips.
- let max_size_bytes = 6 * 1024 * 1024;
+ // 12 MiB over 3 partitions = a 4 MiB share each. The SHARE is what has to
+ // clear the retention floor of one sealed segment (`segment_size` plus the
+ // bus message cap), not the topic-wide figure: a cap that looks large
+ // enough before the divisor still leaves every partition floored and
+ // reclaiming nothing.
+ let max_size_bytes = 12 * 1024 * 1024;
let topic = client
.create_topic(
&Identifier::named(STREAM_NAME).unwrap(),
@@ -578,17 +582,26 @@ pub async fn run_fair_size_based_cleanup_multipartition(client: &IggyClient, dat
let payload = make_payload('E');
- // Send 30 messages per partition = ~9.5 MiB total, exceeds 5.4 MiB threshold
+ // A 10-message batch is already past the 1 MiB segment size on its own and
+ // the crossing batch is written whole, so each batch seals exactly one
+ // segment: 7 batches leave ~7 MiB of SEALED bytes per partition plus an
+ // empty active one, well past the 4 MiB share. Batched rather than one
+ // request per message because each request costs a consensus round-trip
+ // plus an fsync.
+ let batches_per_partition = 7u32;
+ let batch_size = 10u32;
+ let messages_per_partition = batches_per_partition * batch_size;
for partition_id in 0..PARTITIONS_COUNT {
- for i in 0..30 {
- let msg_id = partition_id as u128 * 1000 + i as u128;
- let message = IggyMessage::builder()
- .id(msg_id)
- .payload(payload.clone())
- .build()
- .unwrap();
-
- let mut messages = vec![message];
+ for batch in 0..batches_per_partition {
+ let mut messages: Vec<IggyMessage> = (0..batch_size)
+ .map(|i| {
+ IggyMessage::builder()
+ .id(u128::from(partition_id * 1000 + batch * batch_size + i))
+ .payload(payload.clone())
+ .build()
+ .unwrap()
+ })
+ .collect();
client
.send_messages(
&Identifier::named(STREAM_NAME).unwrap(),
@@ -604,7 +617,7 @@ pub async fn run_fair_size_based_cleanup_multipartition(client: &IggyClient, dat
// Wait for cleaner
tokio::time::sleep(CLEANER_BUFFER).await;
- // Verify segments exist for all partitions
+ // Verify every partition kept an active segment and lost its oldest ones.
for partition_id in 0..PARTITIONS_COUNT {
let partition_path = data_path
.join(format!(
@@ -618,6 +631,35 @@ pub async fn run_fair_size_based_cleanup_multipartition(client: &IggyClient, dat
"Partition {} should have at least 1 segment",
partition_id
);
+
+ let polled = client
+ .poll_messages(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ Some(partition_id),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ messages_per_partition,
+ false,
+ )
+ .await
+ .unwrap();
+ let first_offset = polled
+ .messages
+ .first()
+ .map(|m| m.header.offset)
+ .unwrap_or(0);
+
+ // The divisor is what this asserts: ~7 MiB per partition never reaches
+ // the 12 MiB topic-wide figure, so a cap enforced without dividing by
+ // the partition count would delete nothing here.
+ assert!(
+ first_offset > 0,
+ "Partition {} should have lost its oldest messages to the per-partition share, \
+ got first_offset {}",
+ partition_id,
+ first_offset
+ );
}
client
diff --git a/core/integration/tests/server/scenarios/read_during_persistence_scenario.rs b/core/integration/tests/server/scenarios/read_during_persistence_scenario.rs
index c5b726bfc..dee7bf5d8 100644
--- a/core/integration/tests/server/scenarios/read_during_persistence_scenario.rs
+++ b/core/integration/tests/server/scenarios/read_during_persistence_scenario.rs
@@ -23,24 +23,13 @@ use std::time::{Duration, Instant};
const STREAM_NAME: &str = "eventual-consistency-stream";
const TOPIC_NAME: &str = "eventual-consistency-topic";
-/// Test that specifically targets the background message_saver race condition.
+/// Test that a poll served while messages are still in the journal observes
+/// them, rather than seeing an empty journal with the data not yet on disk.
///
-/// Setup:
-/// - HIGH inline persistence thresholds (messages stay in journal during send)
-/// - SHORT message_saver interval (background flushes happen frequently)
-///
-/// The race occurs when:
-/// 1. Messages are in journal
-/// 2. Background saver calls commit() - journal becomes EMPTY
-/// 3. Background saver starts async disk write
-/// 4. Poll arrives - sees empty journal, data not yet on disk
-#[iggy_harness(server(
- message_saver.interval = "100ms",
- message_saver.enabled = true
-))]
-async fn should_read_messages_during_background_saver_flush(
- harness: &integration::harness::TestHarness,
-) {
+/// Setup: HIGH inline persistence thresholds, so messages stay in the journal
+/// for the whole send and the poll has to read them from there.
+#[iggy_harness]
+async fn should_read_messages_still_in_the_journal(harness: &integration::harness::TestHarness) {
let producer = harness.tcp_root_client().await.unwrap();
producer.create_stream(STREAM_NAME).await.unwrap();
@@ -49,10 +38,10 @@ async fn should_read_messages_during_background_saver_flush(
&Identifier::named(STREAM_NAME).unwrap(),
TOPIC_NAME,
// Both inline thresholds sit far past anything the loop below
- // sends, so nothing is ever flushed on the send path and only the
- // background saver moves data to disk -- which is the race under
- // test. Both have to move: the byte threshold defaults to 1 MiB and
- // would flush on its own long before the message count trips.
+ // sends, so no send ever flushes and every batch is still in the
+ // journal when the poll that follows it runs. Both have to move:
+ // the byte threshold defaults to 1 MiB and would flush on its own
+ // long before the message count trips.
&TopicCreateOptions {
partitions_count: Some(1),
message_expiry: Some(IggyExpiry::NeverExpire),
@@ -80,11 +69,11 @@ async fn should_read_messages_during_background_saver_flush(
let mut next_offset = 0u64;
println!(
- "Starting background saver race test: {} msgs/batch, duration: {}s",
+ "Starting journal read race test: {} msgs/batch, duration: {}s",
messages_per_batch,
test_duration.as_secs()
);
- println!("Inline persistence DISABLED (high thresholds), background saver every 100ms");
+ println!("Inline persistence thresholds raised, so every batch stays in the journal");
while start.elapsed() < test_duration {
let base_offset = next_offset;
diff --git a/core/integration/tests/server/scenarios/restart_offset_skip_scenario.rs b/core/integration/tests/server/scenarios/restart_offset_skip_scenario.rs
index 9516fbb6e..f3f557f1f 100644
--- a/core/integration/tests/server/scenarios/restart_offset_skip_scenario.rs
+++ b/core/integration/tests/server/scenarios/restart_offset_skip_scenario.rs
@@ -63,8 +63,7 @@ pub async fn run(harness: &mut TestHarness) {
&stream_id,
TOPIC_NAME,
// High flush threshold so post-restart messages accumulate in the
- // journal (which is what exposed the base_offset=0 bug); the
- // message_saver flushes the pre-restart data before the restart.
+ // journal (which is what exposed the base_offset=0 bug).
&TopicCreateOptions {
partitions_count: Some(1),
message_expiry: Some(IggyExpiry::NeverExpire),
diff --git a/core/integration/tests/server/scenarios/segment_rotation_race_scenario.rs b/core/integration/tests/server/scenarios/segment_rotation_race_scenario.rs
index ae90c90a0..fb963ad0a 100644
--- a/core/integration/tests/server/scenarios/segment_rotation_race_scenario.rs
+++ b/core/integration/tests/server/scenarios/segment_rotation_race_scenario.rs
@@ -27,7 +27,6 @@
//! `PAYLOAD_FILLER_LEN`)
//! - 8 concurrent producers (2 per protocol: TCP, HTTP, QUIC, WebSocket)
//! - All producers write to the same partition for maximum lock contention
-//! - Short message_saver interval to add more concurrent persist operations
use iggy::prelude::*;
use iggy_common::TransportProtocol;
diff --git a/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs b/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs
index 6c12759bb..40772c090 100644
--- a/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs
+++ b/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs
@@ -120,8 +120,6 @@ async fn setup_resources(client: &IggyClient) {
#[iggy_harness(server(
heartbeat.enabled = true,
heartbeat.interval = "2s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_handle_stale_client_with_manual_reconnection(
harness: &integration::harness::TestHarness,
@@ -265,8 +263,6 @@ async fn should_handle_stale_client_with_manual_reconnection(
#[iggy_harness(server(
heartbeat.enabled = true,
heartbeat.interval = "2s",
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
))]
async fn should_handle_stale_client_with_auto_reconnection(
harness: &integration::harness::TestHarness,
diff --git a/core/integration/tests/server/specific.rs b/core/integration/tests/server/specific.rs
index 276143049..d8cd8a12d 100644
--- a/core/integration/tests/server/specific.rs
+++ b/core/integration/tests/server/specific.rs
@@ -62,8 +62,6 @@ async fn should_handle_single_message_per_batch_with_delayed_persistence(harness
#[iggy_harness(
test_client_transport = [Tcp, WebSocket, Quic],
server(
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true,
quic.max_idle_timeout = "500s",
quic.keep_alive_interval = "15s"
)
@@ -80,8 +78,6 @@ async fn producer_reconnect_after_server_restart(harness: &mut TestHarness) {
#[iggy_harness(
test_client_transport = [Tcp, WebSocket],
server(
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true,
quic.max_idle_timeout = "500s",
quic.keep_alive_interval = "15s"
)
@@ -120,9 +116,8 @@ async fn consumer_offset_ahead_after_crash(harness: &mut TestHarness) {
/// gaps.
///
/// Config: high messages_required_to_save so post-restart messages accumulate in
-/// the journal (exposing the base_offset=0 bug). message_saver flushes pre-restart
-/// data before the restart.
-#[iggy_harness(server(message_saver.enabled = true, message_saver.interval = "1s"))]
+/// the journal (exposing the base_offset=0 bug).
+#[iggy_harness]
async fn restart_offset_skip(harness: &mut TestHarness) {
restart_offset_skip_scenario::run(harness).await;
}
@@ -135,22 +130,14 @@ async fn restart_offset_skip(harness: &mut TestHarness) {
/// Server configuration:
/// - Smallest segment size a topic may declare (1 MiB), plus a payload sized
/// to keep rotations frequent at that floor (~240 rolls per run)
-/// - Short message_saver interval (1s) to add concurrent persist operations
/// - Small messages_required_to_save (32) to trigger more frequent saves
-/// - cache_indexes = none to trigger clear_active_indexes path
///
/// Test configuration:
/// - 8 producers total (2 per protocol: TCP, HTTP, QUIC, WebSocket)
/// - All producers write to the same partition for maximum lock contention
// Concurrency race test: runs over the three VSR transports (TCP/QUIC/
// WebSocket -- HTTP/REST carries no VSR framing).
-#[iggy_harness(server(
- message_saver.interval = "1s",
- segment.cache_indexes = "none",
- tcp.socket_migration = false,
- tcp.socket.override_defaults = true,
- tcp.socket.nodelay = true
-))]
+#[iggy_harness]
async fn segment_rotation_scenario(harness: &TestHarness) {
segment_rotation_race_scenario::run(harness).await;
}
diff --git a/core/integration/tests/server/stats_vsr.rs b/core/integration/tests/server/stats_vsr.rs
index beaa97831..2144199b3 100644
--- a/core/integration/tests/server/stats_vsr.rs
+++ b/core/integration/tests/server/stats_vsr.rs
@@ -23,8 +23,7 @@ use iggy::prelude::*;
use integration::iggy_harness;
#[iggy_harness(
- test_client_transport = [Tcp],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ test_client_transport = [Tcp]
)]
async fn given_connected_clients_when_getting_stats_should_count_clients(harness: &TestHarness) {
let clients = harness.tcp_root_clients(2).await.expect("tcp root clients");
diff --git a/core/integration/tests/server/topic_admission_vsr.rs b/core/integration/tests/server/topic_admission_vsr.rs
index eb8e2977b..74e51813a 100644
--- a/core/integration/tests/server/topic_admission_vsr.rs
+++ b/core/integration/tests/server/topic_admission_vsr.rs
@@ -21,9 +21,11 @@
//! topic, create partitions and delete partitions alike); a custom
//! `max_topic_size` below the configured segment size denies with
//! `InvalidTopicSize`; `ServerDefault` and `Unlimited` sizes pass. Update
-//! 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
+//! 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
//! 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.
@@ -58,8 +60,7 @@ async fn create_topic_with(
}
#[iggy_harness(
- test_client_transport = [Tcp],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ test_client_transport = [Tcp]
)]
async fn given_out_of_bounds_topic_when_creating_should_reject_typed(harness: &TestHarness) {
let client = harness.tcp_root_client().await.expect("tcp root client");
@@ -115,9 +116,99 @@ async fn given_out_of_bounds_topic_when_creating_should_reject_typed(harness: &T
.expect("unlimited max_topic_size is accepted");
}
+// The topic's own segment size is pinned at create so the floor the
+// assertions straddle is exact rather than the shipped default.
+#[iggy_harness(test_client_transport = [Tcp])]
+async fn given_topic_size_below_segment_size_when_updating_should_reject_typed(
+ harness: &TestHarness,
+) {
+ let client = harness.tcp_root_client().await.expect("tcp root client");
+ client
+ .create_stream("update-bounds-stream")
+ .await
+ .expect("create stream");
+ let stream_id = Identifier::from_str_value("update-bounds-stream").expect("stream identifier");
+ let segment_size = IggyByteSize::from_str("8MiB").expect("byte size");
+ let at_floor = MaxTopicSize::Custom(segment_size);
+ client
+ .create_topic(
+ &stream_id,
+ "update-bounds-topic",
+ &TopicCreateOptions {
+ partitions_count: Some(1),
+ message_expiry: Some(IggyExpiry::NeverExpire),
+ max_topic_size: Some(at_floor),
+ segment_size: Some(segment_size),
+ ..TopicCreateOptions::default()
+ },
+ )
+ .await
+ .expect("a topic exactly one segment large is accepted");
+ let topic_id = Identifier::from_str_value("update-bounds-topic").expect("topic identifier");
+
+ let update_to = |max_topic_size: MaxTopicSize| {
+ let client = &client;
+ let stream_id = &stream_id;
+ let topic_id = &topic_id;
+ async move {
+ client
+ .update_topic(
+ stream_id,
+ topic_id,
+ "update-bounds-topic",
+ &TopicUpdateOptions {
+ max_topic_size: Some(max_topic_size),
+ ..TopicUpdateOptions::default()
+ },
+ )
+ .await
+ }
+ };
+
+ let below_floor = MaxTopicSize::Custom(IggyByteSize::from_str("4MiB").expect("byte size"));
+ let invalid_size = IggyError::InvalidTopicSize(below_floor, segment_size).as_code();
+ let result = update_to(below_floor).await;
+ assert!(
+ matches!(&result, Err(error) if error.as_code() == invalid_size),
+ "an update below the segment size must deny with InvalidTopicSize, got {result:?}"
+ );
+ let topic = client
+ .get_topic(&stream_id, &topic_id)
+ .await
+ .expect("get topic")
+ .expect("topic exists");
+ assert_eq!(
+ topic.max_topic_size, at_floor,
+ "the rejected update must not reach consensus, so the stored cap is unchanged"
+ );
+
+ update_to(at_floor)
+ .await
+ .expect("an update to exactly one segment is accepted");
+ update_to(MaxTopicSize::Unlimited)
+ .await
+ .expect("unlimited is exempt from the floor");
+
+ // The sentinel encodes as 0, which the option parser reads as "not set", so
+ // no cap reaches admission at all: the accept proves nothing on its own.
+ // What it must leave behind is the value the previous update stored.
+ update_to(MaxTopicSize::ServerDefault)
+ .await
+ .expect("server default is exempt from the floor");
+ let topic = client
+ .get_topic(&stream_id, &topic_id)
+ .await
+ .expect("get topic")
+ .expect("topic exists");
+ assert_eq!(
+ topic.max_topic_size,
+ MaxTopicSize::Unlimited,
+ "a sentinel update carries no cap, so the stored one survives"
+ );
+}
+
#[iggy_harness(
- test_client_transport = [Tcp],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ test_client_transport = [Tcp]
)]
async fn given_updated_topic_when_getting_topic_should_echo_stored_values(harness: &TestHarness) {
let client = harness.tcp_root_client().await.expect("tcp root client");
@@ -189,66 +280,7 @@ async fn given_updated_topic_when_getting_topic_should_echo_stored_values(harnes
);
}
-#[iggy_harness(
- test_client_transport = [Tcp],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
-)]
-async fn given_update_below_one_segment_when_updating_topic_should_reject_typed(
- harness: &TestHarness,
-) {
- let client = harness.tcp_root_client().await.expect("tcp root client");
- client
- .create_stream("update-bounds-stream")
- .await
- .expect("create stream");
- let stream_id = Identifier::from_str_value("update-bounds-stream").expect("stream identifier");
- create_topic_with(
- &client,
- &stream_id,
- "update-bounds-topic",
- 1,
- MaxTopicSize::Unlimited,
- )
- .await
- .expect("create topic");
- let topic_id = Identifier::from_str_value("update-bounds-topic").expect("topic identifier");
-
- // Create refuses a cap under one segment; an update has to refuse the same
- // value, or the stored map reports a size the topic can never enforce.
- let tiny = MaxTopicSize::Custom(IggyByteSize::from_str("10KiB").expect("byte size"));
- let result = client
- .update_topic(
- &stream_id,
- &topic_id,
- "update-bounds-topic",
- &TopicUpdateOptions {
- max_topic_size: Some(tiny),
- ..TopicUpdateOptions::default()
- },
- )
- .await;
- let invalid_size = IggyError::InvalidTopicSize(tiny, IggyByteSize::default()).as_code();
- assert!(
- matches!(&result, Err(error) if error.as_code() == invalid_size),
- "update to a cap below one segment must deny with InvalidTopicSize, got {result:?}"
- );
-
- let topic = client
- .get_topic(&stream_id, &topic_id)
- .await
- .expect("get topic")
- .expect("topic exists");
- assert_eq!(
- topic.max_topic_size,
- MaxTopicSize::Unlimited,
- "a denied update must leave the stored cap alone"
- );
-}
-
-#[iggy_harness(
- test_client_transport = [Tcp],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
-)]
+#[iggy_harness(test_client_transport = [Tcp])]
async fn given_sentinel_option_when_creating_topic_should_store_the_resolved_default(
harness: &TestHarness,
) {
@@ -301,10 +333,7 @@ async fn given_sentinel_option_when_creating_topic_should_store_the_resolved_def
);
}
-#[iggy_harness(
- test_client_transport = [Tcp],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
-)]
+#[iggy_harness(test_client_transport = [Tcp])]
async fn given_out_of_bounds_partitions_count_when_mutating_should_reject_typed(
harness: &TestHarness,
) {
@@ -373,8 +402,7 @@ async fn given_out_of_bounds_partitions_count_when_mutating_should_reject_typed(
}
#[iggy_harness(
- test_client_transport = [Tcp],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ test_client_transport = [Tcp]
)]
async fn given_over_count_when_deleting_partitions_should_reject_invalid_partitions_count(
harness: &TestHarness,
@@ -436,8 +464,7 @@ async fn given_over_count_when_deleting_partitions_should_reject_invalid_partiti
}
#[iggy_harness(
- test_client_transport = [Tcp],
- server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+ test_client_transport = [Tcp]
)]
async fn given_missing_stream_when_listing_topics_should_return_empty(harness: &TestHarness) {
let client = harness.tcp_root_client().await.expect("tcp root client");
diff --git a/core/message_bus/src/config.rs b/core/message_bus/src/config.rs
index c31d14672..be060b333 100644
--- a/core/message_bus/src/config.rs
+++ b/core/message_bus/src/config.rs
@@ -63,19 +63,16 @@ use std::time::Duration;
///
/// Hardcoded knobs the bus does NOT expose: `max_concurrent_uni_streams = 0`
/// and the CUBIC congestion controller. Both are architectural
-/// invariants of the SDK-client plane (single bidi stream per peer,
-/// no datagram or unidirectional traffic).
+/// invariants of the SDK-client plane (no datagram or unidirectional
+/// traffic).
#[derive(Debug, Clone)]
pub struct QuicTuning {
/// Maximum number of concurrent bidirectional streams per
- /// connection. The bus opens exactly one per peer; setting this
- /// above 1 just preallocates unused quinn-proto state.
+ /// connection. Each SDK command opens a fresh bidi stream, so
+ /// this caps how many commands one client connection may have
+ /// in flight; 1 disables pipelining.
pub max_concurrent_bidi_streams: u32,
- /// Buffer size handed to quinn's outbound datagram queue, in
- /// bytes.
- pub datagram_send_buffer_size: usize,
-
/// Initial path MTU advertised to the peer, in bytes.
pub initial_mtu: u16,
@@ -85,6 +82,13 @@ pub struct QuicTuning {
/// Receive-flow control window per connection, in bytes.
pub receive_window: u32,
+ /// Receive-flow control window per stream, in bytes. Never above
+ /// [`Self::receive_window`], and equal to it under the shipped
+ /// single-stream default. Setting it strictly below is what keeps
+ /// one unread stream from pinning the whole connection window, so
+ /// it only buys anything once several streams share a connection.
+ pub stream_receive_window: u32,
+
/// Interval between QUIC keep-alive PINGs. `Duration::ZERO`
/// disables keep-alive; the connection then relies entirely on
/// [`Self::max_idle_timeout`] for liveness.
@@ -239,11 +243,11 @@ fn build_quic_tuning(quic: &configs::quic::QuicConfig) -> QuicTuning {
QuicTuning {
max_concurrent_bidi_streams: u32::try_from(quic.max_concurrent_bidi_streams)
.unwrap_or(u32::MAX),
- datagram_send_buffer_size: usize::try_from(quic.datagram_send_buffer_size.as_bytes_u64())
- .unwrap_or(usize::MAX),
initial_mtu: u16::try_from(quic.initial_mtu.as_bytes_u64()).unwrap_or(u16::MAX),
send_window: quic.send_window.as_bytes_u64(),
receive_window: u32::try_from(quic.receive_window.as_bytes_u64()).unwrap_or(u32::MAX),
+ stream_receive_window: u32::try_from(quic.stream_receive_window.as_bytes_u64())
+ .unwrap_or(u32::MAX),
keep_alive_interval: quic.keep_alive_interval.get_duration(),
max_idle_timeout: quic.max_idle_timeout.get_duration(),
}
@@ -251,9 +255,9 @@ fn build_quic_tuning(quic: &configs::quic::QuicConfig) -> QuicTuning {
impl Default for QuicTuning {
/// Mirrors the `[quic]` defaults in
- /// `core/server/config.toml`: 64 MiB send/receive windows,
- /// 30 s idle timeout, 10 s keep-alive, 8 KiB initial MTU, 100 KiB
- /// datagram send buffer, single bidi stream per peer.
+ /// `core/server/config.toml`: 64 MiB send/receive windows, a per-stream
+ /// window equal to the connection receive window, 30 s idle timeout, 10 s
+ /// keep-alive, 1200 B initial MTU, one in-flight command per connection.
///
/// Intended for tests and direct callers; production builds
/// derive the field from [`ServerConfig`] so the values stay in
@@ -261,10 +265,10 @@ impl Default for QuicTuning {
fn default() -> Self {
Self {
max_concurrent_bidi_streams: 1,
- datagram_send_buffer_size: 100 * 1024,
- initial_mtu: 8 * 1024,
+ initial_mtu: 1200,
send_window: 64 * 1024 * 1024,
receive_window: 64 * 1024 * 1024,
+ stream_receive_window: 64 * 1024 * 1024,
keep_alive_interval: Duration::from_secs(10),
max_idle_timeout: Duration::from_secs(30),
}
@@ -334,11 +338,6 @@ mod tests {
schema_quic.max_concurrent_bidi_streams,
literal.max_concurrent_bidi_streams
);
- assert_eq!(
- schema_quic.datagram_send_buffer_size, literal.datagram_send_buffer_size,
- "schema datagram_send_buffer_size {} bytes vs literal {} bytes",
- schema_quic.datagram_send_buffer_size, literal.datagram_send_buffer_size
- );
assert_eq!(
schema_quic.initial_mtu, literal.initial_mtu,
"schema initial_mtu {} bytes vs literal {} bytes",
@@ -346,6 +345,10 @@ mod tests {
);
assert_eq!(schema_quic.send_window, literal.send_window);
assert_eq!(schema_quic.receive_window, literal.receive_window);
+ assert_eq!(
+ schema_quic.stream_receive_window,
+ literal.stream_receive_window
+ );
assert_eq!(schema_quic.keep_alive_interval, literal.keep_alive_interval);
assert_eq!(schema_quic.max_idle_timeout, literal.max_idle_timeout);
}
diff --git a/core/message_bus/src/transports/quic.rs b/core/message_bus/src/transports/quic.rs
index df571c994..728c632b0 100644
--- a/core/message_bus/src/transports/quic.rs
+++ b/core/message_bus/src/transports/quic.rs
@@ -119,9 +119,9 @@ const DEFAULT_HANDSHAKE_GRACE: Duration = Duration::from_secs(10);
/// schema (so a misconfigured operator config cannot break the
/// SDK-client plane):
///
-/// - Exactly one bidirectional stream per peer carries every consensus
-/// frame; `tuning.max_concurrent_bidi_streams` clamps to a `u32`
-/// `VarInt` at conversion time but the bus opens only one stream.
+/// - Each SDK request opens a fresh bidirectional stream;
+/// `tuning.max_concurrent_bidi_streams` caps how many a client
+/// connection may hold open at once (clamped to a `u32` `VarInt`).
/// - Zero unidirectional streams (no datagram traffic on this plane).
/// - CUBIC congestion controller (no scheme-level switch yet exposed
/// to operators).
@@ -133,19 +133,18 @@ const DEFAULT_HANDSHAKE_GRACE: Duration = Duration::from_secs(10);
/// # Panics
///
/// Panics if `tuning.max_idle_timeout` is non-zero but outside
-/// quinn-proto's `IdleTimeout` range (`TryFrom<Duration>` rejects
-/// values above `VarInt::MAX_U64` ms, i.e. ~292 years). Schema
-/// validation rejects values that large.
+/// quinn-proto's `IdleTimeout` range: `TryFrom<Duration>` counts the
+/// duration in milliseconds and rejects anything above `VarInt::MAX`
+/// (`2^62 - 1`). `QuicConfig::validate` rejects those values at boot.
#[must_use]
pub fn transport_config_from(tuning: &QuicTuning) -> compio_quic::TransportConfig {
let mut cfg = compio_quic::TransportConfig::default();
cfg.max_concurrent_bidi_streams(VarInt::from_u32(tuning.max_concurrent_bidi_streams))
.max_concurrent_uni_streams(VarInt::from_u32(0))
- .stream_receive_window(VarInt::from_u32(tuning.receive_window))
+ .stream_receive_window(VarInt::from_u32(tuning.stream_receive_window))
.receive_window(VarInt::from_u32(tuning.receive_window))
.send_window(tuning.send_window)
.initial_mtu(tuning.initial_mtu)
- .datagram_send_buffer_size(tuning.datagram_send_buffer_size)
.congestion_controller_factory(Arc::new(congestion::CubicConfig::default()));
if !tuning.max_idle_timeout.is_zero() {
diff --git a/core/message_bus/src/transports/tls/mod.rs b/core/message_bus/src/transports/tls/mod.rs
index 888f9358e..62a2a85fd 100644
--- a/core/message_bus/src/transports/tls/mod.rs
+++ b/core/message_bus/src/transports/tls/mod.rs
@@ -105,7 +105,18 @@ impl std::fmt::Debug for TlsServerCredentials {
/// certificate file contains no certificates, or the key file
/// contains no recognized private key.
pub fn load_pem(cert_path: &Path, key_path: &Path) -> io::Result<TlsServerCredentials> {
- let cert_file = std::fs::File::open(cert_path)?;
+ // Name the path in the message: callers wrap this in an error that only
+ // says which listener failed, so a bare NotFound leaves an operator who
+ // left the path empty with nothing to act on.
+ let cert_file = std::fs::File::open(cert_path).map_err(|error| {
+ io::Error::new(
+ error.kind(),
+ format!(
+ "cannot open certificate file '{}': {error}",
+ cert_path.display()
+ ),
+ )
+ })?;
let mut cert_reader = BufReader::new(cert_file);
let cert_chain: Vec<CertificateDer<'static>> =
rustls_pemfile::certs(&mut cert_reader).collect::<Result<_, _>>()?;
@@ -116,7 +127,15 @@ pub fn load_pem(cert_path: &Path, key_path: &Path) -> io::Result<TlsServerCreden
));
}
- let key_file = std::fs::File::open(key_path)?;
+ let key_file = std::fs::File::open(key_path).map_err(|error| {
+ io::Error::new(
+ error.kind(),
+ format!(
+ "cannot open private key file '{}': {error}",
+ key_path.display()
+ ),
+ )
+ })?;
let mut key_reader = BufReader::new(key_file);
let key_der = rustls_pemfile::private_key(&mut key_reader)?.ok_or_else(|| {
io::Error::new(
@@ -336,6 +355,16 @@ mod tests {
assert_eq!(err.kind(), io::ErrorKind::InvalidData);
}
+ #[test]
+ fn load_pem_names_an_empty_certificate_path() {
+ let err = load_pem(Path::new(""), Path::new("")).expect_err("must fail");
+ assert_eq!(err.kind(), io::ErrorKind::NotFound);
+ assert!(
+ err.to_string().contains("certificate file ''"),
+ "an unset path must be visible in the message, got: {err}"
+ );
+ }
+
#[test]
fn load_pem_rejects_missing_key() {
let cert = rcgen::generate_simple_self_signed(vec!["localhost".to_owned()]).expect("rcgen");
diff --git a/core/metadata/src/stm/stream.rs b/core/metadata/src/stm/stream.rs
index e1a7b3572..eb99ca652 100644
--- a/core/metadata/src/stm/stream.rs
+++ b/core/metadata/src/stm/stream.rs
@@ -48,9 +48,13 @@ use iggy_binary_protocol::requests::partitions::{
use iggy_binary_protocol::requests::streams::{
CreateStreamRequest, DeleteStreamRequest, PurgeStreamRequest, UpdateStreamRequest,
};
+// Only the slab-seeding helpers build a bare `CreateTopicRequest`; without
+// their cfg the import is dead and `-p <crate>` clippy (which skips the
+// simulator feature) rejects it.
+#[cfg(any(test, feature = "simulator"))]
+use iggy_binary_protocol::requests::topics::CreateTopicRequest;
use iggy_binary_protocol::requests::topics::{
- CreateTopicRequest, CreateTopicWithAssignmentsRequest, DeleteTopicRequest, PurgeTopicRequest,
- UpdateTopicRequest,
+ CreateTopicWithAssignmentsRequest, DeleteTopicRequest, PurgeTopicRequest, UpdateTopicRequest,
};
use iggy_binary_protocol::responses::consumer_groups::consumer_group_response::ConsumerGroupResponse;
use iggy_binary_protocol::responses::consumer_groups::get_consumer_group::{
@@ -864,25 +868,54 @@ impl Streams {
}
/// Retention policy for a topic: `(message_expiry, max_topic_size,
- /// partition_count)`, or `None` if the stream or topic is unknown. The
- /// per-shard segment cleaner reads this off-pump to decide local segment
- /// deletion; `partition_count` lets it derive a per-partition size budget.
+ /// partition_count, segment_size)`, or `None` if the stream or topic is
+ /// unknown. The per-shard segment cleaner reads this off-pump to decide
+ /// local segment deletion; `partition_count` lets it derive a
+ /// per-partition size budget and `segment_size` (`None` when the topic
+ /// left the key to the node default) the floor that budget cannot go
+ /// under.
#[must_use]
pub fn topic_retention_config(
&self,
stream_id: usize,
topic_id: usize,
- ) -> Option<(IggyExpiry, MaxTopicSize, usize)> {
+ ) -> Option<(IggyExpiry, MaxTopicSize, usize, Option<IggyByteSize>)> {
self.inner.read(|inner| {
let topic = inner.items.get(stream_id)?.topics.get(topic_id)?;
Some((
topic.message_expiry,
topic.max_topic_size,
topic.partitions.len(),
+ TopicRuntimeOptions::from_resource_options(&topic.options).segment_size,
))
})
}
+ /// A topic's current partition count, or `None` if the stream or topic is
+ /// unknown. Update admission reads it to size the floor a `max_topic_size`
+ /// has to clear across the whole topic; the update request itself carries
+ /// no partitions count.
+ #[must_use]
+ pub fn topic_partitions_count(
+ &self,
+ stream_id: &WireIdentifier,
+ topic_id: &WireIdentifier,
+ ) -> Option<usize> {
+ self.inner.read(|inner| {
+ let stream_slab = inner.resolve_stream_id(stream_id)?;
+ let topic_slab = inner.resolve_topic_id(stream_slab, topic_id)?;
+ Some(
+ inner
+ .items
+ .get(stream_slab)?
+ .topics
+ .get(topic_slab)?
+ .partitions
+ .len(),
+ )
+ })
+ }
+
/// A topic's explicitly set segment size, or `None` when the stream or
/// topic is unknown or the topic left the key to the node default.
///
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index afcb560e8..c3bd2a698 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -92,7 +92,6 @@ use tracing::{debug, warn};
// `SendMessages` retries are at-least-once and may commit multiple times.
// Duplicate suppression is a consensus-layer concern: the VSR client table
// dedups by request id (at-most-once), so the data plane needs no message-id set.
-// (message-id based) if they care.
pub struct IggyPartition<B = IggyMessageBus, SB = PingPongSuperblock>
where
B: MessageBus,
@@ -2893,10 +2892,13 @@ where
}
return Ok(());
}
- // Persist the prefix in segment-sized chunks: a segment seals exactly
- // when its committed bytes reach `max_size`, no matter how many
- // entries this flush happens to cover. A backup commits in bursts
- // behind the primary, so any grouping- or timing-sensitive roll rule
+ // Persist the prefix in segment-sized chunks: a segment seals on the
+ // first flush whose committed bytes reach OR EXCEED `max_size`, no
+ // matter how many entries this flush happens to cover. The batch that
+ // crosses the cap is appended whole, so a sealed segment lands
+ // anywhere in `[max_size, max_size + one maximum batch)` and never
+ // exactly at `max_size`. A backup commits in bursts behind the
+ // primary, so any grouping- or timing-sensitive roll rule
// (like keying rotation on the journal-position `is_full` above)
// seals segments at per-replica offsets, and the offset-keyed segment
// GC staged by the reconciler never converges across the cluster.
@@ -3844,19 +3846,22 @@ where
}
/// Time-expiry plus size-retention in one pass: remove the leading sealed
- /// segments that have expired or that push the partition past `max_bytes`.
- /// Returns the `(segments, messages)` removed.
+ /// segments that have expired or that push the partition's SEALED bytes
+ /// past `max_bytes`. Capped per call by
+ /// `SEGMENT_REMOVAL_BUDGET_PER_PASS`; the returned
+ /// [`SegmentRemoval::budget_spent`] tells the caller whether the rest is
+ /// still waiting.
pub async fn clean_expired_segments(
&mut self,
now: IggyTimestamp,
message_expiry: IggyExpiry,
max_bytes: Option<u64>,
- ) -> (u64, u64) {
+ ) -> SegmentRemoval {
let expired = leading_expired_end(self.log.segments(), now, message_expiry);
let oversized =
max_bytes.and_then(|max_bytes| leading_oversized_end(self.log.segments(), max_bytes));
let Some(up_to) = expired.into_iter().chain(oversized).max() else {
- return (0, 0);
+ return SegmentRemoval::default();
};
self.remove_sealed_segments_up_to(up_to).await
}
@@ -3865,12 +3870,20 @@ where
/// never the active segment and never past the consumer barrier (the
/// minimum committed consumer/group offset). Unlinks the messages and
/// index files and decrements partition stats. Idempotent: an offset below
- /// the oldest sealed segment removes nothing. Returns the
- /// `(segments, messages)` removed.
+ /// the oldest sealed segment removes nothing.
+ ///
+ /// NOT exhaustive: at most `SEGMENT_REMOVAL_BUDGET_PER_PASS` segments go
+ /// per call, so a caller enforcing a retention decision has to re-issue it
+ /// until the layout converges rather than assume one call finished the job.
+ /// Both callers already do: the segment cleaner re-stages on
+ /// [`SegmentRemoval::budget_spent`] and again on its
+ /// `data_maintenance.messages.interval` tick, and the partition reconciler
+ /// re-stages a committed delete watermark on every pass while the first
+ /// local segment still starts below it.
///
/// Holds `write_lock` to serialize against the commit/rotate path, which
/// runs on the separate consensus-tick loop.
- pub async fn remove_sealed_segments_up_to(&mut self, up_to_offset: u64) -> (u64, u64) {
+ pub async fn remove_sealed_segments_up_to(&mut self, up_to_offset: u64) -> SegmentRemoval {
let write_lock = self.write_lock.clone();
let _guard = write_lock.lock().await;
@@ -3880,7 +3893,13 @@ where
let segments = self.log.segments();
let last_idx = segments.len().saturating_sub(1);
let mut removable = 0usize;
- for (idx, segment) in segments.iter().enumerate() {
+ // One past the budget: the extra slot separates a run that ends
+ // exactly on the budget from one with more still waiting.
+ for (idx, segment) in segments
+ .iter()
+ .enumerate()
+ .take(SEGMENT_REMOVAL_BUDGET_PER_PASS + 1)
+ {
if idx == last_idx || !segment.sealed || segment.end_offset > up_to_offset {
break;
}
@@ -3905,8 +3924,12 @@ where
removable
};
- let mut deleted_segments = 0u64;
- let mut deleted_messages = 0u64;
+ let budget_spent = removable > SEGMENT_REMOVAL_BUDGET_PER_PASS;
+ let removable = removable.min(SEGMENT_REMOVAL_BUDGET_PER_PASS);
+ let mut removal = SegmentRemoval {
+ budget_spent,
+ ..SegmentRemoval::default()
+ };
for _ in 0..removable {
// The removable run is always a prefix (oldest first), so the next
// victim is the front once the previous one is gone.
@@ -3946,8 +3969,8 @@ where
self.stats.decrement_segments_count(1);
self.stats.decrement_messages_count(messages_in_segment);
- deleted_segments += 1;
- deleted_messages += messages_in_segment;
+ removal.segments += 1;
+ removal.messages += messages_in_segment;
debug!(
target: "iggy.partitions.diag",
@@ -3959,7 +3982,7 @@ where
);
}
- (deleted_segments, deleted_messages)
+ removal
}
/// Build and install a fresh empty segment starting at `start_offset` with
@@ -4910,6 +4933,41 @@ fn accumulate_committed_info(
info.max_timestamp = info.max_timestamp.max(base_timestamp);
}
+/// Sealed segments one call to [`IggyPartition::remove_sealed_segments_up_to`]
+/// may unlink before it stops and leaves the rest to the next pass.
+///
+/// The removal loop runs inside ONE frame body on the shard pump, and this
+/// shard's consensus ticks are a sibling select arm that stays unpolled while
+/// that body awaits, so the budget is really a bound on how long every OTHER
+/// group on this core goes without a heartbeat. Uncapped it is a bound on the
+/// backlog instead: the first pass after the cleaner is switched on walks
+/// however many segments retention accumulated, which on a large log silences
+/// those groups long enough to lose them to a view change.
+///
+/// A SEGMENT budget standing in for a time bound, so the margin is filesystem
+/// specific: 16 segments is at most 32 unlinks (log plus index), a few hundred
+/// milliseconds on commodity `NVMe` against the shipped 5 s
+/// `cluster.heartbeat_timeout`, and still under 2 s if each unlink costs a
+/// pathological 50 ms on a contended journal. Large enough that steady-state
+/// retention, which reclaims a handful of segments per interval, never reaches
+/// it -- only a backlog does, and that one drains over several passes.
+const SEGMENT_REMOVAL_BUDGET_PER_PASS: usize = 16;
+
+/// What one call to [`IggyPartition::remove_sealed_segments_up_to`] reclaimed.
+///
+/// `budget_spent` reports that the pass stopped on
+/// `SEGMENT_REMOVAL_BUDGET_PER_PASS` rather than on the end of the removable
+/// run, so a caller can re-stage immediately instead of leaving the rest until
+/// its next interval tick. The budget itself stays private: the signal is what
+/// callers need, and reading the number would invite them to rebuild the
+/// comparison.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct SegmentRemoval {
+ pub segments: u64,
+ pub messages: u64,
+ pub budget_spent: bool,
+}
+
/// Highest `end_offset` among the leading run of expired sealed segments, or
/// `None` when none are expired. The last element is the active segment and is
/// never considered. `expiry` must be resolved; a `ServerDefault` expires
@@ -4930,14 +4988,21 @@ fn leading_expired_end(
up_to
}
-/// Highest `end_offset` to drop so the resident size falls to `max_bytes`, or
-/// `None` when already under budget. The active segment (last element) is
-/// never dropped. The budget is per-partition: the cluster has no single owner
-/// of a topic-wide total, so each replica trims its own log.
+/// Highest `end_offset` to drop so the SEALED resident size falls to
+/// `max_bytes`, or `None` when already under budget. The active segment (last
+/// element) is never dropped. The budget is per-partition: the cluster has no
+/// single owner of a topic-wide total, so each replica trims its own log.
+///
+/// The active segment's bytes are excluded from the running total, not merely
+/// from the deletions. Counting bytes that can never be reclaimed lets them
+/// evict the sealed history instead: at a budget near one segment the active
+/// one alone exceeds it, and every sealed segment is dropped no matter how
+/// small the log is.
fn leading_oversized_end(segments: &[Segment], max_bytes: u64) -> Option<u64> {
let last_idx = segments.len().saturating_sub(1);
let mut resident: u64 = segments
.iter()
+ .take(last_idx)
.map(|segment| segment.size.as_bytes_u64())
.sum();
let mut up_to = None;
@@ -7008,6 +7073,68 @@ mod retention_tests {
IggyExpiry::ExpireDuration(IggyDuration::from(Duration::from_secs(1)))
}
+ /// Shape of one sealed segment in the [`partition_with_sealed_run`] fixture.
+ const MESSAGES_PER_SEGMENT: u64 = 10;
+ const BYTES_PER_SEGMENT: u64 = 100;
+
+ /// A topic's configured segment size, and the largest batch that can be
+ /// appended to it.
+ const SEGMENT_SIZE: u64 = 1_000;
+ const MAX_BATCH_SIZE: u64 = 100;
+ /// The size a sealed segment actually reaches. The batch that crosses
+ /// `SEGMENT_SIZE` is appended whole, so a sealed segment closes somewhere
+ /// in `[SEGMENT_SIZE, SEGMENT_SIZE + MAX_BATCH_SIZE)`. Retention asserted
+ /// at exactly `SEGMENT_SIZE` would miss that entirely.
+ const SEALED_SIZE: u64 = SEGMENT_SIZE + 40;
+ /// What the cleaner enforces for a cap of one segment: the per-partition
+ /// share, floored at the largest a sealed segment can be.
+ const FLOORED_BUDGET: u64 = SEGMENT_SIZE + MAX_BATCH_SIZE;
+
+ fn sealed_run_segment(start_offset: u64) -> Segment {
+ let mut segment = segment(
+ start_offset + MESSAGES_PER_SEGMENT - 1,
+ 1,
+ BYTES_PER_SEGMENT,
+ true,
+ );
+ segment.start_offset = start_offset;
+ segment
+ }
+
+ /// Partition whose log is `sealed_count` sealed segments followed by the
+ /// active one, with stats seeded to match. Every storage is the in-memory
+ /// default (no reader, so no path), which is what lets retirement run its
+ /// full body here without touching the filesystem.
+ fn partition_with_sealed_run(sealed_count: u64) -> IggyPartition<IggyMessageBus> {
+ let mut partition = super::tests::test_partition();
+ // The fixture ships one unsealed segment: rewrite it as the head of the
+ // run and append a fresh active segment last, where the removal walk
+ // always stops.
+ partition.log.segments_mut()[0] = sealed_run_segment(0);
+ for index in 1..sealed_count {
+ partition.log.add_persisted_segment(
+ sealed_run_segment(index * MESSAGES_PER_SEGMENT),
+ SegmentStorage::default(),
+ None,
+ None,
+ );
+ }
+ partition.log.add_persisted_segment(
+ Segment::new(
+ sealed_count * MESSAGES_PER_SEGMENT,
+ IggyByteSize::from(0u64),
+ ),
+ SegmentStorage::default(),
+ None,
+ None,
+ );
+ let stats = &partition.stats;
+ stats.increment_segments_count(u32::try_from(sealed_count).expect("run fits a u32"));
+ stats.increment_messages_count(sealed_count * MESSAGES_PER_SEGMENT);
+ stats.increment_size_bytes(sealed_count * BYTES_PER_SEGMENT);
+ partition
+ }
+
#[test]
fn leading_expired_end_skips_active_and_returns_last_expired() {
let segments = vec![
@@ -7055,15 +7182,15 @@ mod retention_tests {
#[test]
fn leading_oversized_end_trims_oldest_until_under_budget() {
- // 4 x 100 = 400 resident, active excluded. Budget 250: drop seg0 (300
- // left) then seg1 (200 <= 250, stop). up_to = seg1.end_offset.
+ // 3 x 100 = 300 SEALED resident, the active segment's bytes excluded.
+ // Budget 250: drop seg0 (200 <= 250, stop). up_to = seg0.end_offset.
let segments = vec![
segment(9, 1, 100, true),
segment(19, 2, 100, true),
segment(29, 3, 100, true),
segment(39, 0, 100, false),
];
- assert_eq!(leading_oversized_end(&segments, 250), Some(19));
+ assert_eq!(leading_oversized_end(&segments, 250), Some(9));
}
#[test]
@@ -7078,6 +7205,38 @@ mod retention_tests {
assert_eq!(leading_oversized_end(&segments, 10), None);
}
+ #[test]
+ fn leading_oversized_end_retains_an_overshot_sealed_segment_at_the_floored_budget() {
+ // The shape admission accepts at the floor: max_topic_size == one
+ // segment. The sealed segment overshot, and the active one holds far
+ // more than the budget. Counting the active segment, or dividing the
+ // cap without a floor, deletes the only history this partition has.
+ let segments = vec![
+ segment(9, 1, SEALED_SIZE, true),
+ segment(19, 0, SEGMENT_SIZE * 5, false),
+ ];
+ assert_eq!(leading_oversized_end(&segments, FLOORED_BUDGET), None);
+ // The same segments against the UNFLOORED share, which is what a cap of
+ // one segment divides into. It is under what a sealed segment reaches,
+ // so the history goes -- which is why the budget carries a floor.
+ assert_eq!(leading_oversized_end(&segments, SEGMENT_SIZE), Some(9));
+ }
+
+ #[test]
+ fn leading_oversized_end_still_drops_the_oldest_once_two_sealed_segments_exceed_the_budget() {
+ // Same budget, one sealed segment more: the cap is real, not disabled.
+ let segments = vec![
+ segment(9, 1, SEALED_SIZE, true),
+ segment(19, 2, SEALED_SIZE, true),
+ segment(29, 0, SEGMENT_SIZE * 5, false),
+ ];
+ assert_eq!(
+ leading_oversized_end(&segments, FLOORED_BUDGET),
+ Some(9),
+ "the oldest sealed segment goes, the newest one stays"
+ );
+ }
+
#[test]
fn nth_oldest_sealed_end_resolves_count_to_offset() {
let segments = vec![
@@ -7109,6 +7268,73 @@ mod retention_tests {
let segments = vec![segment(9, 1, 100, false)];
assert_eq!(nth_oldest_sealed_end(&segments, 1), None);
}
+
+ #[compio::test]
+ async fn removal_spends_the_per_pass_budget_and_resumes_on_the_next_pass() {
+ let budget = u64::try_from(SEGMENT_REMOVAL_BUDGET_PER_PASS).expect("budget fits a u64");
+ let sealed_count = budget + 3;
+ let mut partition = partition_with_sealed_run(sealed_count);
+ // The whole run qualifies: every sealed segment ends at or below
+ // `up_to`, and a partition nobody has committed against has no barrier.
+ let up_to = sealed_count * MESSAGES_PER_SEGMENT - 1;
+ let segments_len = |partition: &IggyPartition<IggyMessageBus>| {
+ u64::try_from(partition.log.segments().len()).expect("log fits a u64")
+ };
+
+ let removal = partition.remove_sealed_segments_up_to(up_to).await;
+ assert_eq!(removal.segments, budget, "one pass stops at the budget");
+ assert_eq!(removal.messages, budget * MESSAGES_PER_SEGMENT);
+ assert!(
+ removal.budget_spent,
+ "a pass that stopped on the budget must ask to be re-staged"
+ );
+ assert_eq!(segments_len(&partition), sealed_count + 1 - budget);
+ assert_eq!(
+ partition.log.segments()[0].start_offset,
+ budget * MESSAGES_PER_SEGMENT,
+ "the surviving run starts where the budget stopped"
+ );
+
+ let remainder = sealed_count - budget;
+ let removal = partition.remove_sealed_segments_up_to(up_to).await;
+ assert_eq!(
+ removal.segments, remainder,
+ "a later pass finishes the run it was handed"
+ );
+ assert_eq!(removal.messages, remainder * MESSAGES_PER_SEGMENT);
+ assert!(
+ !removal.budget_spent,
+ "a pass that drained the run must not re-stage"
+ );
+ assert_eq!(
+ segments_len(&partition),
+ 1,
+ "only the active segment survives"
+ );
+
+ let stats = &partition.stats;
+ assert_eq!(stats.segments_count_inconsistent(), 1);
+ assert_eq!(stats.messages_count_inconsistent(), 0);
+ assert_eq!(stats.size_bytes_inconsistent(), 0);
+ assert_eq!(
+ partition.remove_sealed_segments_up_to(up_to).await,
+ SegmentRemoval::default(),
+ "a converged partition removes nothing"
+ );
+
+ // A run that ends exactly on the budget is finished by the pass that
+ // spends it; re-staging would hand the pump a no-op frame.
+ let mut exact = partition_with_sealed_run(budget);
+ let removal = exact
+ .remove_sealed_segments_up_to(budget * MESSAGES_PER_SEGMENT - 1)
+ .await;
+ assert_eq!(removal.segments, budget, "the whole run goes in one pass");
+ assert!(
+ !removal.budget_spent,
+ "a run that ends on the budget has nothing left to re-stage"
+ );
+ assert_eq!(segments_len(&exact), 1, "only the active segment survives");
+ }
}
#[cfg(test)]
diff --git a/core/partitions/src/lib.rs b/core/partitions/src/lib.rs
index 2b62402cf..81078baa1 100644
--- a/core/partitions/src/lib.rs
+++ b/core/partitions/src/lib.rs
@@ -36,7 +36,7 @@ use iggy_common::IggyError;
pub use iggy_index::IggyIndex;
pub use iggy_index_reader::IggyIndexReader;
pub use iggy_index_writer::IggyIndexWriter;
-pub use iggy_partition::{IggyPartition, PurgeError};
+pub use iggy_partition::{IggyPartition, PurgeError, SegmentRemoval};
pub use iggy_partitions::IggyPartitions;
pub use journal::{EVICTED_RING_BYTES_MAX, EVICTED_RING_CAPACITY};
pub use messages_writer::MessagesWriter;
diff --git a/core/sdk/Cargo.toml b/core/sdk/Cargo.toml
index ea76718f3..4ff5bdcf7 100644
--- a/core/sdk/Cargo.toml
+++ b/core/sdk/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy"
-version = "0.11.0-edge.2"
+version = "0.11.0-edge.3"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
rust-version.workspace = true
diff --git a/core/sdk/src/http/http_client.rs b/core/sdk/src/http/http_client.rs
index f0c751019..7ad665289 100644
--- a/core/sdk/src/http/http_client.rs
+++ b/core/sdk/src/http/http_client.rs
@@ -36,7 +36,6 @@ use std::sync::Arc;
const PUBLIC_PATHS: &[&str] = &[
"/",
- "/metrics",
"/ping",
"/users/login",
"/users/refresh-token",
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index 31c8efd79..db597e52f 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "server"
-version = "0.9.0-edge.3"
+version = "0.9.0-edge.4"
edition = "2024"
license = "Apache-2.0"
publish = false
diff --git a/core/server/config.toml b/core/server/config.toml
index 6b2c423f7..9168ccf26 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -19,14 +19,11 @@
[consumer_group]
# Maximum time a partition can remain in pending revocation before being force-transferred to the target member.
rebalancing_timeout = "30s"
-# How often the periodic checker scans for timed-out pending revocations.
-# TODO(hubcio): inert in the server, which paces the scan from
-# system.sharding.reconcile_periodic_interval instead. Boot warns when set.
-rebalancing_check_interval = "5s"
[data_maintenance.messages]
-# Enables or disables the expired message cleaner process.
-cleaner_enabled = false
+# Enables or disables the segment cleaner. It deletes the oldest sealed segments of
+# topics with a finite message_expiry or max_topic_size, per partition, best-effort.
+cleaner_enabled = true
# Interval for running the message cleaner.
interval = "1 m"
@@ -172,6 +169,8 @@ use_base64_secret = false
enabled = true
# Specifies the endpoint for accessing metrics, e.g., "/metrics".
+# This route requires authentication like every other read: scrapers must
+# present a bearer credential (JWT or personal access token).
endpoint = "/metrics"
# TLS (Transport Layer Security) configuration for HTTP.
@@ -198,15 +197,6 @@ enabled = true
# For example, "127.0.0.1:8090" listens on localhost only on port 8090.
address = "127.0.0.1:8090"
-# Enable TCP socket migration across shards.
-# TODO(hubcio): inert in the server, not implemented. Boot warns when set.
-socket_migration = true
-
-# Whether to use ipv4 or ipv6
-# TODO(hubcio): inert in the server, which takes the family from the
-# tcp.address string. Boot warns when set.
-ipv6 = false
-
# TLS configuration for the TCP server.
[tcp.tls]
# Enables or disables TLS for TCP connections.
@@ -214,40 +204,20 @@ ipv6 = false
# `false` leaves TCP connections unencrypted.
enabled = false
-# Enables or disables self-signed certificate generation.
-# `true` generates a self-signed certificate if cert files don't exist.
-# `false` requires certificate files to exist at the specified paths.
+# Certificate source for the TCP TLS listener.
+# `true` generates an ephemeral loopback certificate, but only while cert_file
+# does not exist on disk: an existing PEM pair is loaded instead (the server
+# logs which way it went).
+# `false` loads cert_file / key_file, which must both exist.
self_signed = true
-# Path to the TLS certificate file.
+# Path to the TLS certificate file. Loaded when the file exists or
+# self_signed = false.
cert_file = "core/certs/iggy_cert.pem"
-# Path to the TLS key file.
+# Path to the TLS key file. Read together with cert_file.
key_file = "core/certs/iggy_key.pem"
-# Configuration for the TCP socket
-# TODO(hubcio): the whole section is inert in the server, which leaves the OS
-# defaults in place. Boot warns when override_defaults is set.
-[tcp.socket]
-# Whether to overwrite the OS-default socket parameters
-override_defaults = false
-
-# SO_RCVBUF: maximum size of the receive buffer, can be clamped by the OS
-recv_buffer_size = "100 KB"
-
-# SO_SNDBUF: maximum size of the send buffer, can be clamped by the OS
-send_buffer_size = "100 KB"
-
-# SO_KEEPALIVE: whether to regularly send a keepalive packet maintaining the connection
-keepalive = false
-
-# TCP_NODELAY: enable/disable the Nagle algorithm which buffers data before sending segments
-nodelay = false
-
-# SO_LINGER: delay to wait for while data is being transmitted before closing the socket after a
-# close or shutdown call has been received
-linger = "0 s"
-
# QUIC protocol configuration.
[quic]
# Controls whether the QUIC server is enabled.
@@ -259,29 +229,42 @@ enabled = true
# For example, "127.0.0.1:8080" binds to localhost on port 8080.
address = "127.0.0.1:8080"
-# Maximum number of simultaneous bidirectional streams per QUIC
-# connection. The message bus opens exactly one bidi stream per peer
-# (no multiplexing), so any value above 1 is wasted on extra
-# preallocated quinn-proto state.
+# How many bidirectional streams a client may keep open at once on a single
+# QUIC connection. The SDK opens a fresh stream per command, so 1 caps a
+# connection at one in-flight command: commands do not pipeline, and the next
+# one waits for the current reply. The known cost is one stream-credit round
+# trip between back-to-back requests on the same connection. Raising it is a
+# transport decision only: the server's accept loop takes one bidi at a time
+# and waits for its reply before accepting the next, which is what keeps
+# partition operations (they all share the connection's current request id)
+# distinguishable, and a higher cap does not make the handlers concurrent.
max_concurrent_bidi_streams = 1
-# Size of the buffer for sending datagrams in QUIC. Binary-aligned to
-# match QuicTuning::default() (102_400 bytes) and the rest of this [quic]
-# block, which uses MiB throughout (`send_window`, `receive_window`).
-datagram_send_buffer_size = "100 KiB"
-
-# Initial Maximum Transmission Unit (MTU) for QUIC connections. Binary
-# units for the same reason as `datagram_send_buffer_size`.
-initial_mtu = "8 KiB"
-
-# Send-flow window per connection. Sized to fit a single max-size
-# framed message (`message_bus.max_message_size`) without
-# head-of-line wait.
+# Initial path MTU for QUIC connections, before MTU discovery has run.
+# 1200 bytes is the QUIC minimum and quinn's own default: safe on every
+# path, and MTU discovery probes upward from it automatically. Values
+# above the real path MTU cause packet loss until black-hole detection
+# resets the connection back to 1200.
+initial_mtu = "1200 B"
+
+# Server-side flow-control budget for unacknowledged data the server may have
+# in flight on one connection. A ceiling, not a reservation, and rarely the
+# operative limit: the receiver's own window binds first, and the Rust SDK's
+# QUIC client sets both of its windows to about 100 KB.
send_window = "64 MiB"
-# Receive-flow window per connection. Symmetric with `send_window`.
+# Server-side flow-control budget for data the server will buffer from one
+# connection, shared by all of its streams. Symmetric with `send_window`.
receive_window = "64 MiB"
+# Per-stream slice of `receive_window`. Equal to it, because
+# `max_concurrent_bidi_streams` is 1: with a single stream per connection a
+# smaller carve-out protects nothing, it only caps the one stream early.
+# Lower it below `receive_window` if that cap is ever raised, so one unread
+# stream cannot pin the whole connection window. Must not exceed
+# `receive_window`; equality is allowed.
+stream_receive_window = "64 MiB"
+
# Interval for sending QUIC keep-alive PINGs. One third of
# `max_idle_timeout` so up to two consecutive losses fit before the
# idle timer closes the connection. Set to "0 s" to disable.
@@ -293,50 +276,19 @@ max_idle_timeout = "30 s"
# QUIC certificate configuration.
[quic.certificate]
-# Indicates whether the QUIC certificate is self-signed.
-# `true` for self-signed certificates, often used in internal or testing environments.
-# `false` for certificates issued by a certificate authority, common in production.
+# Certificate source for the QUIC listener.
+# `true` always generates an ephemeral loopback certificate and ignores
+# cert_file / key_file even when those files exist on disk; the server logs a
+# warning naming the files it skipped.
+# `false` loads cert_file / key_file, which must both exist.
self_signed = true
-# Path to the QUIC TLS certificate file.
-cert_file = "core/certs/iggy_cert.pem"
-
-# Path to the QUIC TLS key file.
-key_file = "core/certs/iggy_key.pem"
-
-# Configuration for the QUIC socket
-# TODO(hubcio): the whole section is inert in the server, which leaves the OS
-# defaults in place. Boot warns when override_defaults is set.
-[quic.socket]
-# Whether to override the OS-default socket parameters
-override_defaults = false
-
-# SO_RCVBUF: maximum size of the receive buffer, can be clamped by the OS
-recv_buffer_size = "64 KB"
-
-# SO_SNDBUF: maximum size of the send buffer, can be clamped by the OS
-send_buffer_size = "64 KB"
-
-# SO_KEEPALIVE: whether to regularly send a keepalive packet maintaining the connection
-keepalive = false
-
-# Message saver configuration.
-[message_saver]
-# Enables or disables the background process for saving buffered data to disk.
-# `true` ensures data is periodically written to disk.
-# `false` turns off automatic saving, relying on other triggers for data persistence.
-enabled = true
-
-# Controls whether data saving is synchronous (enforce fsync) or asynchronous.
-# `true` for synchronous saving, ensuring data integrity at the cost of performance.
-# `false` for asynchronous saving, improving performance but with delayed data writing.
-# TODO(hubcio): inert in the server, which only flushes on shutdown and has no
-# periodic saver to configure. Boot warns when set.
-enforce_fsync = true
+# Path to the QUIC TLS certificate file. Required when self_signed = false,
+# left empty here so the shipped self-signed default has nothing to skip.
+cert_file = ""
-# Interval for running the message saver.
-# TODO(hubcio): inert in the server, see enforce_fsync above. Boot warns when set.
-interval = "30 s"
+# Path to the QUIC TLS key file. Required when self_signed = false.
+key_file = ""
# Personal access token configuration.
[personal_access_token]
@@ -355,10 +307,25 @@ interval = "1 m"
# Heartbeat configuration
[heartbeat]
-# Enables or disables the client heartbeat verification process.
+# Enables or disables the client heartbeat verification process. When
+# enabled, a connection that sends nothing (no request, no PING) for
+# 1.2 x interval has its session released so its consumer groups
+# rebalance off it. Only a connection that still holds a group
+# membership is evicted; one that holds none is left alone and is
+# reaped when its socket closes.
+# TODO(hubcio): flip the default to true once every SDK heartbeats
+# automatically; today the blocking Java and C# SDKs never ping, so an
+# enabled default would evict their idle consumer-group members.
enabled = false
-# Interval for expected client heartbeats
-interval = "5 s"
+# Interval for expected client heartbeats. The Rust, Go, Python, Node and
+# async Java SDKs ping automatically every 5 s, well inside the resulting 36 s
+# staleness window, but only from a connected high-level client: the Rust and
+# async Java pingers are armed by connect(), so a session that logs in without
+# it never pings. The blocking Java and the C# SDK have no automatic heartbeat
+# at all, only a manual ping. Wherever nothing pings, an idle consumer-group
+# member is evicted, and only the application can keep it alive (ping) or
+# bring it back afterwards (reconnect).
+interval = "30 s"
# OpenTelemetry configuration
[telemetry]
@@ -386,35 +353,6 @@ endpoint = "http://localhost:7281/v1/traces"
# Base path for system data storage.
path = "local_data"
-# Backup configuration
-# TODO(hubcio): backup is not supported; both paths below are
-# inert. Boot warns when either is set.
-[system.backup]
-# Path for storing backup.
-path = "backup"
-
-# Compatibility conversion configuration
-[system.backup.compatibility]
-# Subpath of the backup directory where converted segment data is stored after compatibility conversion.
-path = "compatibility"
-
-# TODO(hubcio): the three tunables below are inert in the server, whose state
-# writes do not go through the legacy retrying file layer. Boot warns when any
-# is set.
-[system.state]
-# Determines whether to enforce file synchronization on state updates (boolean).
-# `true` ensures immediate writing of data to disk for durability.
-# `false` allows the OS to manage write operations, which can improve performance.
-enforce_fsync = false
-
-# Maximum number of retries for a failed file operation (e.g., append, overwrite).
-# This defines how many times the system will attempt the operation before failing.
-max_file_operation_retries = 1
-
-# Delay between retries in case of a failed file operation.
-# This helps to avoid immediate repeated attempts and can reduce load.
-retry_delay = "1 s"
-
# Runtime configuration.
[system.runtime]
# Path for storing runtime data.
@@ -456,11 +394,6 @@ rotation_check_interval = "1 h"
# Time to retain log files before deletion. Avoid less than 1s, too.
retention = "7 days"
-# Interval for printing system information to the log.
-# TODO(hubcio): inert in the server, which has no sysinfo printer. Boot warns
-# when set.
-sysinfo_print_interval = "10 s"
-
# Encryption configuration
[system.encryption]
# Determines whether server-side data encryption for the messages payloads and state commands is enabled (boolean).
@@ -473,19 +406,6 @@ enabled = false
# This key is required and used only if encryption is enabled.
key = ""
-# Compression configuration
-[system.compression]
-# Allows overriding the default compression algorithm per data segment (boolean).
-# `true` permits different compression algorithms for individual segments.
-# `false` means all data segments use the default compression algorithm.
-# TODO(hubcio): inert in the server, where live compression is already per-topic
-# from the request. Boot warns when set.
-allow_override = false
-
-# The default compression algorithm used for data storage (string).
-# "none" indicates no compression, other values can specify different algorithms.
-default_algorithm = "none"
-
# Stream configuration
[system.stream]
# Path for storing stream-related data (string).
@@ -540,15 +460,7 @@ validate_checksum = true
# Unsupported: setting this to `true` aborts boot.
archive_expired = false
-# Controls whether to cache indexes (time and positional) for segment access.
-# Possible values:
-# - "true" or "all": keeps indexes in memory, speeding up data retrieval at the cost of memory
-# - "open_segment": keeps indexes in memory only for the currently open segment
-# - "false" or "none": reads indexes from disk, which can conserve memory at the cost of access speed
-# TODO(hubcio): inert in the server, which picks its own index residency. Boot
-# warns when set.
-cache_indexes = "open_segment"
-
+# Recovery configuration in case of lost data
[system.recovery]
# Controls whether streams/topics/partitions should be recreated if the expected data for existing state is missing (boolean).
# Unsupported: setting this to `true` aborts boot.
@@ -583,8 +495,17 @@ enabled = false
# Unique cluster name (string).
# All nodes in the same cluster must share the same name.
# This prevents accidental cross-cluster communication.
+# Permanent on-disk identity: the name is hashed into the cluster id stamped
+# into every metadata and partition superblock on first boot. Changing it later
+# makes the server refuse to start against existing data, since the recovered id
+# no longer matches the configured one. Renaming means starting from an empty
+# data directory.
name = "iggy-cluster"
+# Consensus runs on a fixed 10ms tick, so every duration below is converted to
+# whole ticks: values are rounded down to a multiple of 10ms, and anything under
+# 10ms is raised to a single tick rather than firing sooner.
+
# Backup-side liveness window for a consensus plane's primary (duration).
# A replica that sees no primary traffic for this long starts a view change.
# Raise it on oversubscribed hosts where scheduling stalls fake primary
@@ -689,6 +610,10 @@ key_file = ""
# PEM trust anchor(s) the dialer verifies peer certificates against.
# Unused when self_signed = true.
+# The dialer verifies each peer against the 'ip' string from that peer's
+# cluster.nodes entry, so a roster of literal IPs needs peer certificates
+# carrying matching IP SANs. Hostname-only certificates fail verification;
+# use hostnames in the roster if the certificates only have DNS SANs.
ca_file = ""
# Full roster of cluster members. Byte-identical on every node. The running
@@ -874,7 +799,15 @@ accept_unmasked_frames = false
[websocket.tls]
enabled = false
+
+# Certificate source for the WSS listener.
+# `true` generates an ephemeral loopback certificate, but only while cert_file
+# does not exist on disk: an existing PEM pair is loaded instead (the server
+# logs which way it went).
+# `false` loads cert_file / key_file, which must both exist.
self_signed = true
+
+# Loaded when the file exists or self_signed = false.
cert_file = "core/certs/iggy_cert.pem"
key_file = "core/certs/iggy_key.pem"
@@ -936,14 +869,15 @@ evicted_ring_bytes_max = "16 MiB"
# state-transfer chunk requests. PER SHARD, and shard count defaults to core
# count, so the process-wide high-water is this times the core count on top of
# page cache -- keep that product in mind before raising it. The default is a
-# FIXED 2176 MiB: two sealed segments at the SHIPPED system.segment.size of
-# 1 GiB, each of which can close one whole message_bus.max_message_size past
-# its target, which is why it is not 2 GiB. It does not track your segment
-# size. How many groups this shard serves at once IS derived from yours:
-# floor(this / max(partition.transfer_artifact_bytes_max,
-# system.segment.size + 64 MiB)), minimum one. So raising either that knob or
-# system.segment.size without raising this lowers concurrency and can take it
-# to one, serialising rejoins, and nothing at boot warns about it.
+# FIXED 2176 MiB: two sealed segments at the built-in 1 GiB segment ceiling,
+# each of which can close one whole message_bus.max_message_size past its
+# target, which is why it is not 2 GiB. That ceiling is a compile-time
+# constant: no config key moves it, and a per-topic segment_size is refused
+# above it. How many groups this shard serves at once is
+# floor(this / max(partition.transfer_artifact_bytes_max, 1 GiB + 64 MiB)),
+# minimum one. So raising transfer_artifact_bytes_max without raising this
+# lowers concurrency and can take it to one, serialising rejoins; boot warns
+# when it drops below two.
# Below one segment a single rejoining node thrashes the cache by itself and
# every miss re-reads and re-hashes a whole segment to serve one 256 KiB chunk.
# Running under the budget costs re-reads, not failures.
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index 475a8b36c..8b1613577 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -3649,6 +3649,7 @@ fn load_replica_tls_ctx(
};
let credentials = if tls.self_signed {
+ warn_ignored_certificate_files("cluster.tls", &tls.cert_file, &tls.key_file);
let san = config
.cluster
.nodes
@@ -3729,7 +3730,7 @@ fn load_tcp_tls_server_credentials(
config: &ServerConfig,
) -> Result<TlsServerCredentials, ServerError> {
let tls = &config.tcp.tls;
- if tls.self_signed && !Path::new(&tls.cert_file).exists() {
+ if ephemeral_certificate("tcp.tls", tls.self_signed, &tls.cert_file) {
return Ok(self_signed_for_loopback());
}
@@ -3783,7 +3784,7 @@ async fn start_websocket_listener(
fn load_wss_server_credentials(config: &ServerConfig) -> Result<TlsServerCredentials, ServerError> {
let tls = &config.websocket.tls;
- if tls.self_signed && !Path::new(&tls.cert_file).exists() {
+ if ephemeral_certificate("websocket.tls", tls.self_signed, &tls.cert_file) {
return Ok(self_signed_for_loopback());
}
@@ -3800,6 +3801,11 @@ fn load_quic_server_credentials(
) -> Result<replica_io::QuicServerCredentials, ServerError> {
let certificate = &config.quic.certificate;
if certificate.self_signed {
+ warn_ignored_certificate_files(
+ "quic.certificate",
+ &certificate.cert_file,
+ &certificate.key_file,
+ );
let (cert_chain, key_der) = server_common::generate_self_signed_certificate("localhost")
.map_err(|error| ServerError::ListenerCredentials {
transport: "quic",
@@ -3825,6 +3831,44 @@ fn load_quic_server_credentials(
})
}
+/// Client-listener certificate precedence: `self_signed = true` mints an
+/// ephemeral loopback certificate only while `cert_file` is absent from disk.
+/// An existing PEM pair wins, so a deployment that lays certificates down
+/// serves them without also having to unset the flag - the contract every
+/// SDK test lane relies on when it points the server at `core/certs/`.
+fn ephemeral_certificate(section: &str, self_signed: bool, cert_file: &str) -> bool {
+ if !self_signed {
+ return false;
+ }
+ if Path::new(cert_file).exists() {
+ info!(
+ "{section}.self_signed = true but cert_file = {cert_file} exists on disk; loading it - remove the file or clear the path to serve an ephemeral certificate"
+ );
+ return false;
+ }
+ true
+}
+
+/// `self_signed = true` never reads the PEM pair (cluster and QUIC keep the
+/// flag authoritative: their generated certificates carry non-loopback SANs),
+/// so a cert path resolving on disk looks active to an operator who never
+/// asked for it.
+fn warn_ignored_certificate_files(section: &str, cert_file: &str, key_file: &str) {
+ let found: Vec<String> = [("cert_file", cert_file), ("key_file", key_file)]
+ .into_iter()
+ .filter(|(_, path)| Path::new(path).exists())
+ .map(|(field, path)| format!("{field} = {path}"))
+ .collect();
+ if found.is_empty() {
+ return;
+ }
+
+ warn!(
+ "{section}.self_signed = true, ignoring certificate files found on disk ({}); set {section}.self_signed = false to load them",
+ found.join(", ")
+ );
+}
+
fn parse_socket_addr(context: &'static str, address: &str) -> Result<SocketAddr, ServerError> {
address
.parse()
diff --git a/core/server/src/dispatch.rs b/core/server/src/dispatch.rs
index 8ad3c146c..4600ca0dd 100644
--- a/core/server/src/dispatch.rs
+++ b/core/server/src/dispatch.rs
@@ -46,6 +46,7 @@ use crate::responses::{
connected_client_to_response, current_metadata_commit, resolve_partition_namespace,
resolve_partition_request_namespace,
};
+use crate::segment_cleaner::UNENFORCEABLE_TOPIC_SIZE_WARN;
use crate::session_manager::SessionManager;
use crate::snapshot;
use crate::users::maybe_rewrite_user_password_request;
@@ -110,6 +111,7 @@ use metadata::impls::metadata::{
build_truncate_partition_client_message_with_identifiers,
};
use metadata::permissioner::Permissioner;
+use metadata::stm::stream::Streams;
use partitions::{AutoCommitApplied, PollPlan, PollingArgs, PollingConsumer};
use secrecy::ExposeSecret;
use server_common::Message;
@@ -844,6 +846,75 @@ pub(crate) fn validate_topic_size_floor(
Ok(())
}
+/// Announce an accepted `max_topic_size` the server cannot enforce as written.
+///
+/// [`validate_topic_size_floor`] admits any cap of one segment or more, but
+/// retention runs PER PARTITION and floors each partition's share at one SEALED
+/// segment, which reaches up to one maximum bus frame past `segment_size`. A cap
+/// between the two is stored and echoed back verbatim while the server actually
+/// keeps `(segment_size + max_message_size) * partitions_count`, so the only
+/// moment an operator can be told is the one where they set it.
+///
+/// Warns rather than rejects: which caps are accepted is client-visible wire
+/// behavior, and tightening it would break topics that already exist.
+pub(crate) fn warn_unenforceable_topic_size(
+ max_topic_size: MaxTopicSize,
+ segment_size_bytes: u64,
+ max_message_size_bytes: usize,
+ partitions_count: u32,
+) {
+ let MaxTopicSize::Custom(configured) = max_topic_size else {
+ return;
+ };
+ let max_message_size_bytes = u64::try_from(max_message_size_bytes).unwrap_or(u64::MAX);
+ let per_partition_floor = segment_size_bytes.saturating_add(max_message_size_bytes);
+ let topic_floor = per_partition_floor.saturating_mul(u64::from(partitions_count));
+ if configured.as_bytes_u64() >= topic_floor {
+ return;
+ }
+ warn!(
+ max_topic_size = configured.as_bytes_u64(),
+ partitions_count,
+ segment_size = segment_size_bytes,
+ enforced_per_partition = per_partition_floor,
+ "{UNENFORCEABLE_TOPIC_SIZE_WARN}"
+ );
+}
+
+/// Announce the same unenforceable cap when partitions are ADDED to a topic.
+///
+/// The cap is topic-wide but enforcement is per partition, so every added
+/// partition shrinks the share: a cap that cleared the floor when the topic was
+/// created can stop clearing it here. The request carries only the delta, so
+/// the stored cap, segment size and current partition count come from metadata.
+pub(crate) fn warn_unenforceable_topic_size_on_partition_add(
+ streams: &Streams,
+ stream_id: &WireIdentifier,
+ topic_id: &WireIdentifier,
+ max_message_size_bytes: usize,
+ added_partitions_count: u32,
+) {
+ let Some(((stream_slab, topic_slab), _)) = streams.partition_count_context(stream_id, topic_id)
+ else {
+ return;
+ };
+ let Some((_, max_topic_size, partitions_count, segment_size)) =
+ streams.topic_retention_config(stream_slab, topic_slab)
+ else {
+ return;
+ };
+ warn_unenforceable_topic_size(
+ max_topic_size,
+ segment_size.map_or(iggy_common::DEFAULT_SEGMENT_SIZE, |segment_size| {
+ segment_size.as_bytes_u64()
+ }),
+ max_message_size_bytes,
+ u32::try_from(partitions_count)
+ .unwrap_or(u32::MAX)
+ .saturating_add(added_partitions_count),
+ );
+}
+
/// Reject option keys outside the resource's catalog, pre-consensus. Unknown
/// keys are rejected rather than skipped: a silently ignored knob would hand
/// the client server defaults without it ever learning. Streams and users
@@ -1175,18 +1246,31 @@ async fn handle_client_request<B, MJ, S, SB>(
{
validate_preallocated_topic_bytes(segment_size, create_topic.partitions_count)?;
}
- validate_topic_bounds(
- create_topic.partitions_count,
- options
- .max_topic_size
- .unwrap_or(MaxTopicSize::ServerDefault),
+ let max_topic_size = options
+ .max_topic_size
+ .unwrap_or(MaxTopicSize::ServerDefault);
+ validate_topic_bounds(create_topic.partitions_count, max_topic_size, segment_size)?;
+ warn_unenforceable_topic_size(
+ max_topic_size,
segment_size,
- )
+ shard.bus_max_message_size(),
+ create_topic.partitions_count,
+ );
+ Ok(())
}),
Operation::CreatePartitions => CreatePartitionsRequest::decode_from(request_body(&request))
.map_err(|_| IggyError::InvalidCommand)
.and_then(|create_partitions| {
- validate_partitions_change_count(create_partitions.partitions_count)
+ validate_partitions_change_count(create_partitions.partitions_count)?;
+ let metadata = shard.plane.metadata();
+ warn_unenforceable_topic_size_on_partition_add(
+ metadata.mux_stm.streams(),
+ &create_partitions.stream_id,
+ &create_partitions.topic_id,
+ shard.bus_max_message_size(),
+ create_partitions.partitions_count,
+ );
+ Ok(())
}),
Operation::DeletePartitions => DeletePartitionsRequest::decode_from(request_body(&request))
.map_err(|_| IggyError::InvalidCommand)
@@ -1209,15 +1293,24 @@ async fn handle_client_request<B, MJ, S, SB>(
// topic can never enforce. The floor is this topic's own
// segment size, since that key is create-only.
let metadata = shard.plane.metadata();
- let segment_size = metadata
- .mux_stm
- .streams()
+ let streams = metadata.mux_stm.streams();
+ let segment_size = streams
.topic_segment_size(&update_topic.stream_id, &update_topic.topic_id)
.map_or_else(
|| iggy_common::DEFAULT_SEGMENT_SIZE,
|segment_size| segment_size.as_bytes_u64(),
);
- validate_topic_size_floor(max_topic_size, segment_size)
+ validate_topic_size_floor(max_topic_size, segment_size)?;
+ let partitions_count = streams
+ .topic_partitions_count(&update_topic.stream_id, &update_topic.topic_id)
+ .unwrap_or(0);
+ warn_unenforceable_topic_size(
+ max_topic_size,
+ segment_size,
+ shard.bus_max_message_size(),
+ u32::try_from(partitions_count).unwrap_or(u32::MAX),
+ );
+ Ok(())
}),
Operation::UpdateStream => UpdateStreamRequest::decode_from(request_body(&request))
.map_err(|_| IggyError::InvalidCommand)
@@ -1924,17 +2017,23 @@ pub(crate) async fn run_heartbeat_verifier<B, MJ, S, SB>(
S: 'static,
SB: SuperblockStore + 'static,
{
- // Legacy `MAX_THRESHOLD`: a client is stale once it misses ~1.2 intervals.
- let max_age = interval.mul_f64(1.2);
+ // Legacy `MAX_THRESHOLD`: a client is stale once it misses 1.2 intervals.
+ // Integer 6/5 rather than `mul_f64`, which panics on an absurd interval.
+ let max_age = interval.saturating_mul(6) / 5;
loop {
- if stop_rx.try_recv().is_ok() {
+ // `Ok(_)`: stop signalled -> exit. `Err(_)`: interval elapsed -> pass.
+ // Waiting on the stop channel rather than sleeping past it keeps this
+ // task inside the shutdown drain budget, which is shorter than the
+ // heartbeat interval.
+ let stop_signal = compio::time::timeout(interval, stop_rx.recv()).await;
+ if stop_signal.is_ok() {
break;
}
// Production-only wall clock: the heartbeat verifier is spawned solely
// by `build_shard_for_thread`, never by the simulator's
- // `wire_shell_handlers`, so this read is off the deterministic path. If
- // it is ever driven under the deterministic executor, route it through
- // the bus sleep / injected clock (as the pump tick is) to stay virtual.
+ // `wire_shell_handlers`, so neither the interval wait above nor this
+ // read is on a deterministic path. Driving this task under the
+ // deterministic executor means routing both through the injected clock.
let stale = sessions
.borrow()
.collect_stale(max_age, std::time::Instant::now());
@@ -1963,8 +2062,6 @@ pub(crate) async fn run_heartbeat_verifier<B, MJ, S, SB>(
evict_stale_client(&shard, &sessions, transport_client_id).await;
}
}
-
- shard.bus.sleep(interval).await;
}
}
@@ -3296,9 +3393,9 @@ where
///
/// A resume window becomes worth having once SDK-side identity stability
/// lands, at which point it needs a timer of its own -- riding the heartbeat
-/// verifier is not an option, since that only runs when `heartbeat.enabled`
-/// is set and `collect_stale` keys off the heartbeat interval, so ungating it
-/// would mass-evict consumer-group members on a deployment that does not ping.
+/// verifier would tie the grace period to heartbeat configuration, since
+/// `collect_stale` keys off `heartbeat.interval` and the verifier does not run
+/// at all when `heartbeat.enabled` is false.
/// Deliberately does NOT drop the local `ClientTable` slot first:
/// `submit_logout_*` short-circuits when the slot is already gone, so a
/// pre-emptive local removal would suppress the `Logout` and leave peer
diff --git a/core/server/src/http.rs b/core/server/src/http.rs
index b58498cb9..28c6ebeb0 100644
--- a/core/server/src/http.rs
+++ b/core/server/src/http.rs
@@ -171,7 +171,7 @@ pub async fn start(
state,
max_request_size,
cors,
- metrics_endpoint,
+ metrics_endpoint.as_deref(),
http_config.web_ui,
);
@@ -255,14 +255,14 @@ const PING_PATH: &str = "/ping";
/// the inner layer's `iggy-view`.
///
/// `metrics_endpoint`, present only when `[http.metrics]` is enabled, mounts
-/// the public scrape route among the local routes (a scrape must describe the
-/// serving node, never a forwarded primary) and switches on the
+/// the auth-only scrape route among the local routes (a scrape must describe
+/// the serving node, never a forwarded primary) and switches on the
/// request-counting layer.
fn router(
state: HttpState,
max_request_size: usize,
cors: Option<CorsLayer>,
- metrics_endpoint: Option<String>,
+ metrics_endpoint: Option<&str>,
web_ui: bool,
) -> Router {
// Cloned for the response layer so `Iggy-View` reads the live view per
@@ -307,7 +307,7 @@ fn router(
.route("/cluster/metadata", get(get_cluster_metadata))
.route("/clients", get(get_clients))
.route("/clients/{client_id}", get(get_client));
- let local = match &metrics_endpoint {
+ let local = match metrics_endpoint {
Some(endpoint) => local.route(endpoint, get(metrics::get_metrics)),
None => local,
};
@@ -318,14 +318,13 @@ fn router(
.layer(DefaultBodyLimit::max(max_request_size))
.layer(from_fn(move |request: Request, next: Next| {
let view_source = view_source.clone();
- // `/ping` and the metrics scrape are the success routes reached
- // without proving a credential, so they must not leak the
- // cluster-internal view number (the anon-leak gate). Every other
- // route authenticates before its handler, so a success/redirect
- // there is an authed flow that may carry the header; the login
- // routes prove credentials on success.
- let suppress_view = request.uri().path() == PING_PATH
- || metrics_endpoint.as_deref() == Some(request.uri().path());
+ // `/ping` is the one success route reached without proving a
+ // credential, so it must not leak the cluster-internal view number
+ // (the anon-leak gate). Every other route - the metrics scrape
+ // included - authenticates before its handler, so a success or
+ // redirect there is an authed flow that may carry the header; the
+ // login routes prove credentials on success.
+ let suppress_view = request.uri().path() == PING_PATH;
async move {
let response = next.run(request).await;
if suppress_view {
diff --git a/core/server/src/http/handlers.rs b/core/server/src/http/handlers.rs
index fc3203731..e138249e3 100644
--- a/core/server/src/http/handlers.rs
+++ b/core/server/src/http/handlers.rs
@@ -121,7 +121,8 @@ use shard::{PartitionRead, PartitionReadReply};
use crate::auth::{verify_login_credentials, verify_pat_credentials};
use crate::dispatch::{
resolve_consumer_offset_request, resolve_poll_request, validate_option_keys,
- validate_topic_bounds, validate_topic_size_floor,
+ validate_topic_bounds, validate_topic_size_floor, warn_unenforceable_topic_size,
+ warn_unenforceable_topic_size_on_partition_add,
};
use crate::http::error::{
Consistency, ConsistencyQuery, CustomError, PartitionWriteError, ProduceAck, ProduceQuery,
@@ -884,12 +885,15 @@ pub(in crate::http) async fn create_topic(
validate_preallocated_topic_bytes(segment_size, command.partitions_count)
.map_err(WriteError::Rejected)?;
}
- validate_topic_bounds(
- command.partitions_count,
- parsed.max_topic_size.unwrap_or(MaxTopicSize::ServerDefault),
+ let max_topic_size = parsed.max_topic_size.unwrap_or(MaxTopicSize::ServerDefault);
+ validate_topic_bounds(command.partitions_count, max_topic_size, segment_size)
+ .map_err(WriteError::Rejected)?;
+ warn_unenforceable_topic_size(
+ max_topic_size,
segment_size,
- )
- .map_err(WriteError::Rejected)?;
+ state.shard.bus_max_message_size(),
+ command.partitions_count,
+ );
let request = CreateTopicRequest {
stream_id: identifier_to_wire(&stream_id).map_err(WriteError::Rejected)?,
partitions_count: command.partitions_count,
@@ -941,15 +945,23 @@ pub(in crate::http) async fn update_topic(
.max_topic_size
{
let metadata = state.shard.plane.metadata();
- let segment_size = metadata
- .mux_stm
- .streams()
+ let streams = metadata.mux_stm.streams();
+ let segment_size = streams
.topic_segment_size(&stream_wire, &topic_wire)
.map_or_else(
|| iggy_common::DEFAULT_SEGMENT_SIZE,
|segment_size| segment_size.as_bytes_u64(),
);
validate_topic_size_floor(max_topic_size, segment_size).map_err(WriteError::Rejected)?;
+ let partitions_count = streams
+ .topic_partitions_count(&stream_wire, &topic_wire)
+ .unwrap_or(0);
+ warn_unenforceable_topic_size(
+ max_topic_size,
+ segment_size,
+ state.shard.bus_max_message_size(),
+ u32::try_from(partitions_count).unwrap_or(u32::MAX),
+ );
}
let request = UpdateTopicRequest {
stream_id: stream_wire,
@@ -1035,6 +1047,14 @@ pub(in crate::http) async fn create_partitions(
topic_id: identifier_to_wire(&topic_id).map_err(WriteError::Rejected)?,
partitions_count: command.partitions_count,
};
+ let metadata = state.shard.plane.metadata();
+ warn_unenforceable_topic_size_on_partition_add(
+ metadata.mux_stm.streams(),
+ &request.stream_id,
+ &request.topic_id,
+ state.shard.bus_max_message_size(),
+ request.partitions_count,
+ );
let body = request.to_bytes();
SendWrapper::new(submit_write(
&state,
diff --git a/core/server/src/http/jwt.rs b/core/server/src/http/jwt.rs
index 29fae7e43..2e0781f00 100644
--- a/core/server/src/http/jwt.rs
+++ b/core/server/src/http/jwt.rs
@@ -214,12 +214,14 @@ impl JwtManager {
return self.decode_self_issued(token);
}
- // Try the self-issued verify first. A trusted-issuer token is signed by
- // the issuer's key, which the self-issued verify rejects at the algorithm
- // check (when self-issued is HS*, the shipped default) or at signature
- // verification (if an operator configures self-issued RS*/ES*) - either
- // way it can never accept an external token; a valid self-issued token
- // returns here and the classification below only runs once it has failed.
+ // Try the self-issued verify first. Self-issued is always HS* (config
+ // validation refuses anything else), so a token signed by an issuer's
+ // key fails here either at the algorithm check, when its header names a
+ // non-HMAC alg, or at the signature check, when the header claims HS*
+ // but the signature was not made with the local secret. Either way an
+ // external token can never be accepted as self-issued; a valid
+ // self-issued token returns here and the classification below only runs
+ // once it has failed.
if let Ok(claims) = self.decode_self_issued(token) {
return Ok(claims);
}
@@ -347,7 +349,7 @@ fn normalize_issuer_url(url: &str) -> String {
/// Fill in an unconfigured JWT secret exactly like the legacy HTTP server:
/// both empty -> one random ephemeral secret; one empty -> mirror the other;
-/// both set but different under an HMAC algorithm -> warn (they must match).
+/// both set but different -> warn (they must match).
fn normalize_secrets(mut config: HttpJwtConfig, cluster_psk: Option<&str>) -> HttpJwtConfig {
match (
config.encoding_secret.is_empty(),
@@ -390,11 +392,9 @@ fn normalize_secrets(mut config: HttpJwtConfig, cluster_psk: Option<&str>) -> Ht
config.decoding_secret = config.encoding_secret.clone();
}
(false, false) => {
- if config.encoding_secret != config.decoding_secret
- && config.algorithm.starts_with("HS")
- {
+ if config.encoding_secret != config.decoding_secret {
warn!(
- "JWT encoding and decoding secrets are different but algorithm is {} (HMAC) - both secrets must be identical for symmetric algorithms.",
+ "JWT encoding and decoding secrets are different but self-issued tokens are signed and verified with one shared secret (algorithm is {}) - both secrets must be identical.",
config.algorithm
);
}
diff --git a/core/server/src/http/metrics.rs b/core/server/src/http/metrics.rs
index bb18a8a9b..8261d6dc6 100644
--- a/core/server/src/http/metrics.rs
+++ b/core/server/src/http/metrics.rs
@@ -31,6 +31,7 @@ use prometheus_client::registry::Registry;
use send_wrapper::SendWrapper;
use tracing::error;
+use crate::http::extractor::Identity;
use crate::http::state::HttpState;
/// The legacy server's metric set, registered under the same names and help
@@ -134,8 +135,10 @@ pub(in crate::http) fn validated_endpoint(
}
/// `GET <http.metrics.endpoint>`: the metric set in prometheus text
-/// exposition. Public - reached without proving a credential, exactly like the
-/// legacy endpoint.
+/// exposition. Auth-only, like `/stats`: the `Identity` extractor rejects a
+/// missing or invalid bearer with 401, and any authenticated user may scrape
+/// (no RBAC rule guards it). Scrapers present a JWT or a raw PAT the same way
+/// every read route accepts them.
///
/// The entity gauges sample the same reads `/stats` serves: the metadata STM
/// stream and user maps plus the stats-registry rollups, whose partition-plane
@@ -143,7 +146,10 @@ pub(in crate::http) fn validated_endpoint(
/// in flight. The clients count scatter-gathers the per-shard session managers
/// exactly like `GET /clients` and turns partial when a shard misses the reply
/// deadline.
-pub(in crate::http) async fn get_metrics(State(state): State<HttpState>) -> String {
+pub(in crate::http) async fn get_metrics(
+ State(state): State<HttpState>,
+ _identity: Identity,
+) -> String {
let (streams_count, topics_count, partitions_count, segments_count, messages_count) = state
.shard
.plane
diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs
index fa2d3f735..4c3aeecfb 100644
--- a/core/server/src/partition_reconciler.rs
+++ b/core/server/src/partition_reconciler.rs
@@ -1084,9 +1084,10 @@ fn shards_table_has_epoch(ctx: &ReconcilerCtx, ns: IggyNamespace, epoch: u64) ->
/// so a redundant pass triggered by an unrelated revision bump is harmless.
/// A watermark whose enforcement is still incomplete (first local segment
/// starts below it) counts as pending work: the pump may be blocked by a
-/// consumer barrier or by a rejoin whose offsets arrive via journal repair,
-/// and neither unblocking bumps `Streams::revision`, so the pass must keep
-/// the reconciler ticking until the layout converges.
+/// consumer barrier, by a rejoin whose offsets arrive via journal repair, or
+/// by the per-pass removal budget that keeps one trim from monopolising the
+/// pump, and no such unblocking bumps `Streams::revision`, so the pass must
+/// keep the reconciler ticking until the layout converges.
fn reconcile_segment_truncations(ctx: &ReconcilerCtx, counters: &mut PassCounters) {
let partitions = ctx.shard.plane.partitions();
let namespaces: Vec<_> = partitions.namespaces().copied().collect();
diff --git a/core/server/src/segment_cleaner.rs b/core/server/src/segment_cleaner.rs
index 45795988e..eac91c156 100644
--- a/core/server/src/segment_cleaner.rs
+++ b/core/server/src/segment_cleaner.rs
@@ -31,9 +31,18 @@ use consensus::{MetadataHandle, PartitionsHandle};
use iggy_common::{IggyExpiry, IggyTimestamp, MaxTopicSize};
use metadata::impls::metadata::StreamsFrontend;
use shard::Receiver;
+use std::collections::HashSet;
use std::rc::Rc;
use std::time::Duration;
-use tracing::trace;
+use tracing::{trace, warn};
+
+/// Disclosure for a `max_topic_size` the server cannot enforce as written,
+/// shared by admission (which says it once, when the cap is set) and the
+/// cleaner (which says it once per topic, when it floors the budget).
+pub const UNENFORCEABLE_TOPIC_SIZE_WARN: &str = concat!(
+ "max_topic_size is below what retention can enforce; the segment cleaner keeps at ",
+ "least one sealed segment per partition instead, and enforces nothing while disabled"
+);
/// Run the cleaner until `stop` fires. Wakes every `interval`; expiry and size
/// are evaluated against wall-clock and resident bytes, so no metadata-commit
@@ -44,11 +53,15 @@ pub async fn run_segment_cleaner(shard: Rc<ServerShard>, stop: Receiver<()>, int
interval_ms = interval.as_millis(),
"segment cleaner started"
);
+ // Topic incarnations whose floored budget has already been reported. Only
+ // grows, and only for topics whose configured cap is below what can be
+ // enforced, so it stays far smaller than the topic table.
+ let mut reported_floors = HashSet::new();
loop {
// `Ok(_)`: stop signalled -> exit. `Err(_)`: interval elapsed -> pass.
match compio::time::timeout(interval, stop.recv()).await {
Ok(_) => break,
- Err(_) => stage_owned_partitions(&shard),
+ Err(_) => stage_owned_partitions(&shard, &mut reported_floors),
}
}
trace!(shard = shard.id, "segment cleaner exited");
@@ -57,14 +70,21 @@ pub async fn run_segment_cleaner(shard: Rc<ServerShard>, stop: Receiver<()>, int
/// Stage a cleaner pass for every partition this shard owns whose topic has a
/// retention policy. Reads config off-pump and hands the resolved decision to
/// the pump; partitions with no policy are skipped without a frame.
-fn stage_owned_partitions(shard: &Rc<ServerShard>) {
+///
+/// `reported_floors` carries the topics already named in a floor-raise log
+/// across passes, so the disclosure lands once per topic instead of once per
+/// interval tick. Slab ids are reused after a delete, so the creation revision
+/// is part of the key: a topic recreated into the same slots discloses again.
+fn stage_owned_partitions(
+ shard: &Rc<ServerShard>,
+ reported_floors: &mut HashSet<(usize, usize, u64)>,
+) {
let now = IggyTimestamp::now();
let namespaces: Vec<_> = shard.plane.partitions().namespaces().copied().collect();
let streams = shard.plane.metadata().mux_stm.streams();
- // Resolved once per pass, not per partition: the node default is a `Cell`
- // written at bootstrap and never after.
+ let max_message_size = u64::try_from(shard.bus_max_message_size()).unwrap_or(u64::MAX);
for namespace in namespaces {
- let Some((message_expiry, max_topic_size, partition_count)) =
+ let Some((message_expiry, max_topic_size, partition_count, segment_size)) =
streams.topic_retention_config(namespace.stream_id(), namespace.topic_id())
else {
continue;
@@ -77,24 +97,87 @@ fn stage_owned_partitions(shard: &Rc<ServerShard>) {
message_expiry,
IggyExpiry::NeverExpire | IggyExpiry::ServerDefault
);
- let max_bytes = per_partition_size_budget(
+ // A topic that left `segment_size` unset rotates at the node default,
+ // which is the same value admission floors its cap against.
+ let segment_size = segment_size.map_or(iggy_common::DEFAULT_SEGMENT_SIZE, |segment_size| {
+ segment_size.as_bytes_u64()
+ });
+ let budget = per_partition_size_budget(
max_topic_size,
iggy_common::DEFAULT_MAX_TOPIC_SIZE,
partition_count,
+ segment_size.saturating_add(max_message_size),
);
- if !has_expiry && max_bytes.is_none() {
+ if !has_expiry && budget.is_none() {
continue;
}
- shard.request_clean_partition(namespace, now, message_expiry, max_bytes);
+ if let Some(budget) = budget
+ && budget.max_bytes > budget.configured_share
+ && reported_floors.insert((
+ namespace.stream_id(),
+ namespace.topic_id(),
+ streams
+ .created_revision_for_namespace(namespace)
+ .unwrap_or_default(),
+ ))
+ {
+ warn!(
+ shard = shard.id,
+ stream_id = namespace.stream_id(),
+ topic_id = namespace.topic_id(),
+ max_topic_size = budget.resolved_cap,
+ partition_count,
+ configured_share = budget.configured_share,
+ enforced_per_partition = budget.max_bytes,
+ "{UNENFORCEABLE_TOPIC_SIZE_WARN}"
+ );
+ }
+ shard.request_clean_partition(
+ namespace,
+ now,
+ message_expiry,
+ budget.map(|budget| budget.max_bytes),
+ );
}
}
+/// A topic's enforced per-partition budget, next to the share it was derived
+/// from. The two differ exactly when the configured cap is too small to be
+/// enforceable, which is what the cleaner discloses once per topic.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct PartitionSizeBudget {
+ /// Cap on this partition's SEALED bytes.
+ max_bytes: u64,
+ /// `max_topic_size / partition_count`, before the floor.
+ configured_share: u64,
+ /// The topic-wide cap the share came from, with `ServerDefault` already
+ /// resolved to the node default. The sentinel's own `as_bytes_u64()` is 0,
+ /// so this is what a caller has to report.
+ resolved_cap: u64,
+}
+
/// Per-partition byte budget for a topic, or `None` for "no cap".
///
/// The cluster has no single owner of a topic-wide total, so each partition
/// keeps an equal share.
///
+/// That share is raised to `sealed_segment_ceiling`, the largest a single
+/// SEALED segment can be. Admission rejects a cap below one segment TOPIC-wide,
+/// so an accepted cap can still divide into a share smaller than one segment,
+/// and a share below what one sealed segment reaches retains no history at all:
+/// the oldest sealed segment is over budget the moment it seals, every pass,
+/// forever. The floor makes the cap mean "keep at least the newest sealed
+/// segment", which is the least an enforceable retention policy can say. The
+/// raise is disclosed by the caller, since the enforced value then diverges
+/// from the one `GetTopic` echoes.
+///
+/// The ceiling is `segment_size` plus the CONFIGURED bus frame cap, not a
+/// compile-time constant: rotation fires after the append that crosses the
+/// target, so the crossing batch lands whole and a sealed segment runs up to
+/// one maximum bus frame past `segment_size`. An operator who raises that knob
+/// grows real sealed segments with it.
+///
/// `ServerDefault` is resolved against the node default HERE, at enforcement
/// time. Create admission rewrites the sentinel before replication, but an
/// UPDATE to `ServerDefault` leaves it in committed state, and reading that as
@@ -108,7 +191,8 @@ fn per_partition_size_budget(
max_topic_size: MaxTopicSize,
default_max_topic_size: u64,
partition_count: usize,
-) -> Option<u64> {
+ sealed_segment_ceiling: u64,
+) -> Option<PartitionSizeBudget> {
let resolved = match max_topic_size {
MaxTopicSize::ServerDefault => MaxTopicSize::from(default_max_topic_size),
sized => sized,
@@ -116,7 +200,12 @@ fn per_partition_size_budget(
match resolved {
MaxTopicSize::Custom(size) => {
let divisor = u64::try_from(partition_count).unwrap_or(1).max(1);
- Some(size.as_bytes_u64() / divisor)
+ let configured_share = size.as_bytes_u64() / divisor;
+ Some(PartitionSizeBudget {
+ max_bytes: configured_share.max(sealed_segment_ceiling),
+ configured_share,
+ resolved_cap: size.as_bytes_u64(),
+ })
}
// `From<u64>` maps 0 back to `ServerDefault`, so a node default of 0
// lands here as "no cap" rather than as a trim-everything budget.
@@ -126,9 +215,20 @@ fn per_partition_size_budget(
#[cfg(test)]
mod tests {
- use super::per_partition_size_budget;
+ use super::{PartitionSizeBudget, per_partition_size_budget};
use iggy_common::MaxTopicSize;
+ /// A ceiling low enough that the share is what the assertion is about.
+ const NO_FLOOR: u64 = 1;
+
+ fn budget(max_bytes: u64, configured_share: u64, resolved_cap: u64) -> PartitionSizeBudget {
+ PartitionSizeBudget {
+ max_bytes,
+ configured_share,
+ resolved_cap,
+ }
+ }
+
#[test]
fn server_default_resolves_to_the_node_default_at_enforcement_time() {
// A topic UPDATED back to `ServerDefault` keeps the sentinel in
@@ -136,19 +236,19 @@ mod tests {
// it diverges from a topic CREATED with the same setting (whose
// sentinel admission already rewrote).
assert_eq!(
- per_partition_size_budget(MaxTopicSize::ServerDefault, 4096, 4),
- Some(1024),
+ per_partition_size_budget(MaxTopicSize::ServerDefault, 4096, 4, NO_FLOOR),
+ Some(budget(1024, 1024, 4096)),
"the node default is resolved and split across partitions"
);
// Shipped config: server default is unlimited -> still no cap.
assert_eq!(
- per_partition_size_budget(MaxTopicSize::ServerDefault, u64::MAX, 4),
+ per_partition_size_budget(MaxTopicSize::ServerDefault, u64::MAX, 4, NO_FLOOR),
None
);
// A zero node default must read as "no cap", never as a zero budget
// that trims every sealed segment.
assert_eq!(
- per_partition_size_budget(MaxTopicSize::ServerDefault, 0, 4),
+ per_partition_size_budget(MaxTopicSize::ServerDefault, 0, 4, NO_FLOOR),
None
);
}
@@ -156,17 +256,53 @@ mod tests {
#[test]
fn explicit_sizes_ignore_the_node_default() {
assert_eq!(
- per_partition_size_budget(MaxTopicSize::Custom(4096u64.into()), 64, 2),
- Some(2048)
+ per_partition_size_budget(MaxTopicSize::Custom(4096u64.into()), 64, 2, NO_FLOOR),
+ Some(budget(2048, 2048, 4096))
);
assert_eq!(
- per_partition_size_budget(MaxTopicSize::Unlimited, 64, 2),
+ per_partition_size_budget(MaxTopicSize::Unlimited, 64, 2, NO_FLOOR),
None
);
// Zero partitions must not divide by zero.
assert_eq!(
- per_partition_size_budget(MaxTopicSize::Custom(4096u64.into()), 64, 0),
- Some(4096)
+ per_partition_size_budget(MaxTopicSize::Custom(4096u64.into()), 64, 0, NO_FLOOR),
+ Some(budget(4096, 4096, 4096))
+ );
+ }
+
+ #[test]
+ fn a_share_below_one_sealed_segment_is_raised_to_it() {
+ // 8 MiB segments, 1 MiB bus cap: a sealed segment reaches 9 MiB.
+ let segment_size = 8 * 1024 * 1024;
+ let ceiling = segment_size + 1024 * 1024;
+
+ // The shape admission accepts at the floor: cap == one segment, one
+ // partition. The bare share (8 MiB) is under what a sealed segment
+ // reaches, so it would delete each segment as it seals.
+ assert_eq!(
+ per_partition_size_budget(MaxTopicSize::Custom(segment_size.into()), 0, 1, ceiling),
+ Some(budget(ceiling, segment_size, segment_size)),
+ "an at-floor cap enforces one sealed segment, not zero"
+ );
+
+ // The divisor reintroduces the same shape at any accepted cap: 16 x 8
+ // MiB passes admission, then splits into 8 MiB per partition.
+ assert_eq!(
+ per_partition_size_budget(
+ MaxTopicSize::Custom((segment_size * 16).into()),
+ 0,
+ 16,
+ ceiling
+ ),
+ Some(budget(ceiling, segment_size, segment_size * 16))
+ );
+
+ // A cap with room to spare keeps its share untouched, so the floor
+ // never loosens a policy that was already enforceable.
+ let roomy = ceiling * 10;
+ assert_eq!(
+ per_partition_size_budget(MaxTopicSize::Custom(roomy.into()), 0, 1, ceiling),
+ Some(budget(roomy, roomy, roomy))
);
}
}
diff --git a/core/server/src/session_manager.rs b/core/server/src/session_manager.rs
index 3a896b096..84635a8b4 100644
--- a/core/server/src/session_manager.rs
+++ b/core/server/src/session_manager.rs
@@ -577,11 +577,12 @@ mod tests {
assert_eq!(mgr.get_session(c2), Some((200, 20)));
}
// Every disconnect releases its consensus session, group member or not.
- // Holding the slot open for a resume window instead leaked it: the sweep
- // that would have collected it rides the heartbeat verifier, which only
- // runs when `heartbeat.enabled` is set, and that ships false -- so the
- // slot survived for the process lifetime and pushed the client table
- // toward capacity eviction, which silently erases dedup watermarks.
+ // Holding the slot open for a resume window instead leaked it: nothing
+ // sweeps it afterwards. The heartbeat verifier runs only when
+ // `heartbeat.enabled` is set, and even then evicts only connections that
+ // still hold a consumer-group membership, so the slot survived for the
+ // process lifetime and pushed the client table toward capacity eviction,
+ // which silently erases dedup watermarks.
#[test]
fn disconnect_releases_the_bound_session_for_logout() {
let mut mgr = SessionManager::new();
diff --git a/core/server_common/src/messages_batch_mut.rs b/core/server_common/src/messages_batch_mut.rs
index 902eeda51..bf7bbd10e 100644
--- a/core/server_common/src/messages_batch_mut.rs
+++ b/core/server_common/src/messages_batch_mut.rs
@@ -402,139 +402,6 @@ impl IggyMessagesBatchMut {
IggyMessageView::new(&self.messages[start..end]).ok()
}
- /// This helper function is used to parse newly appended chunks in the `new_buffer`.
- /// The function iterates over the range `[chunk_start..chunk_start + chunk_len]`,
- /// constructing `IggyMessageView` instances to compute message sizes. For each message,
- /// a corresponding index entry is created in `new_indexes`. The `offset_in_new_buffer`
- /// is incremented by each message’s size to preserve the correct offsets for
- /// subsequent messages in the new buffer.
- #[allow(clippy::too_many_arguments)]
- fn rebuild_indexes_for_chunk(
- new_buffer: &PooledBuffer,
- new_indexes: &mut IggyIndexesMut,
- offset_in_new_buffer: &mut u32,
- chunk_start: usize,
- chunk_len: usize,
- ) {
- let chunk_end = chunk_start + chunk_len;
- let mut current = chunk_start;
-
- while current < chunk_end {
- let Ok(view) = IggyMessageView::new(&new_buffer[current..]) else {
- error!("Corrupt message in already-validated chunk at offset {current}");
- break;
- };
- let msg_size = view.size();
- *offset_in_new_buffer += msg_size as u32;
- new_indexes.insert(0, *offset_in_new_buffer, 0);
-
- current += msg_size;
- }
- }
-
- /// Removes messages at the specified indexes and returns a new batch.
- ///
- /// This function efficiently creates a new `IggyMessagesBatchMut` by copying only the
- /// messages that should be kept, and rebuilding the index entries. Note that `put()`
- /// can be memmove underneath due to the way memory is handled.
- ///
- /// # Arguments
- ///
- /// * `indexes_to_remove` - A slice of message indexes (0-based) to remove
- ///
- /// # Returns
- ///
- /// A new `IggyMessagesBatchMut` with the specified messages removed
- pub fn remove_messages(&mut self, indexes_to_remove: &[u32], current_position: u32) {
- /*
- * A temporary list of message boundaries is first collected for each index
- * that should be removed. Chunks of data that are not removed are appended
- * to a new buffer, and indexes are rebuilt to reflect the shifted positions.
- * In this process, split_to() is used to carve out slices from the source
- * buffer, and those slices are either copied or discarded, depending on
- * whether they are part of the messages that are to be removed.
- * This allows for avoiding copying unnecessary data and ensures that indexes
- * match the newly constructed buffer.
- */
- if indexes_to_remove.is_empty() || self.is_empty() {
- return;
- }
-
- let msg_count = self.count() as usize;
- if indexes_to_remove.len() > msg_count {
- return;
- }
-
- let current_size = self.size();
- let mut size_to_remove = 0;
- let boundaries_to_remove: Vec<(usize, usize)> = indexes_to_remove
- .iter()
- .filter_map(|&idx| {
- self.get_message_boundaries(idx as usize)
- .inspect(|boundaries| {
- size_to_remove += (boundaries.1 - boundaries.0) as u32;
- })
- })
- .collect();
-
- assert_eq!(
- boundaries_to_remove.len(),
- indexes_to_remove.len(),
- "Could not retrieve valid boundaries for some message indexes: {indexes_to_remove:?}, boundaries: {boundaries_to_remove:?}"
- );
-
- let new_size = current_size - size_to_remove;
- let new_message_count = msg_count as u32 - indexes_to_remove.len() as u32;
-
- let mut new_buffer = PooledBuffer::with_capacity(new_size as usize);
- let mut new_indexes =
- IggyIndexesMut::with_capacity(new_message_count as usize, current_position);
-
- let mut source = std::mem::take(&mut self.messages);
- let mut last_pos = 0_usize;
- let mut new_pos = current_position;
-
- for &(start, end) in &boundaries_to_remove {
- if start > last_pos {
- let keep_len = start - last_pos;
- let chunk = source.split_to(keep_len);
- let chunk_start_in_new_buffer = new_buffer.len();
- new_buffer.put(chunk);
-
- Self::rebuild_indexes_for_chunk(
- &new_buffer,
- &mut new_indexes,
- &mut new_pos,
- chunk_start_in_new_buffer,
- keep_len,
- );
- }
-
- let removed_message_size = end - start;
- if removed_message_size > 0 {
- let _ = source.split_to(removed_message_size);
- }
-
- last_pos = end;
- }
-
- if !source.is_empty() {
- let chunk_start_in_new_buffer = new_buffer.len();
- let chunk_len = source.len();
- new_buffer.put(source);
- Self::rebuild_indexes_for_chunk(
- &new_buffer,
- &mut new_indexes,
- &mut new_pos,
- chunk_start_in_new_buffer,
- chunk_len,
- );
- }
-
- self.messages = new_buffer;
- self.indexes = new_indexes;
- }
-
/// Validates that all messages in batch have correct checksums.
pub fn validate_checksums(&self) -> Result<(), IggyError> {
for message in self.iter() {
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index bae836f8b..d11ef938b 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -1564,6 +1564,14 @@ where
self.bus_max_message_size.set(max_message_size);
}
+ /// The configured `[message_bus] max_message_size`. Also the largest batch
+ /// the bus will frame, and so the most a sealed segment can overshoot
+ /// `segment_size`: rotation fires after the append that crosses the cap.
+ #[must_use]
+ pub const fn bus_max_message_size(&self) -> usize {
+ self.bus_max_message_size.get()
+ }
+
/// Mint a fresh, never-zero nonce for a register or logout forward.
///
/// `replica` rides the high half, which separates the nonce spaces of
@@ -1955,14 +1963,30 @@ where
max_bytes: Option<u64>,
) {
let Some(sender) = self.senders.get(self.id as usize) else {
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::PARTITION,
+ crate::metrics::frame_drop_reason::UNROUTABLE,
+ );
return;
};
- let _ = sender.try_send(ShardFrame::lifecycle(LifecycleFrame::CleanPartition {
+ // Fire-and-forget: a refused pass is picked up by the cleaner's next
+ // maintenance tick, so the drop only has to be visible, not recovered.
+ if let Err(error) = sender.try_send(ShardFrame::lifecycle(LifecycleFrame::CleanPartition {
namespace,
now,
message_expiry,
max_bytes,
- }));
+ })) {
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::PARTITION,
+ crate::coordinator::classify_try_send_err(&error),
+ );
+ tracing::debug!(
+ shard = self.id,
+ namespace_raw = namespace.inner(),
+ "segment cleaner pass refused by own inbox: {error:?}"
+ );
+ }
}
/// Stage a `TruncatePartition` enforcement for `namespace` on this shard's
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index d93697e9d..7bdee26ff 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -545,10 +545,10 @@ where
// task already resolved the retention decision off-pump, so
// this only mutates, serialized with reads on the same loop.
if let Some(partition) = self.plane.partitions().get_mut_by_ns(&namespace) {
- let (segments, messages) = partition
+ let removal = partition
.clean_expired_segments(now, message_expiry, max_bytes)
.await;
- if segments > 0 {
+ if removal.segments > 0 {
// Any unlink invalidates what this shard is SERVING:
// the offer names files that are gone and the payload
// cache can answer from RAM without touching disk, so a
@@ -560,11 +560,25 @@ where
tracing::debug!(
shard = self.id,
namespace_raw = namespace.inner(),
- segments,
- messages,
+ segments = removal.segments,
+ messages = removal.messages,
"segment cleaner removed sealed segments"
);
}
+ if removal.budget_spent {
+ // The pass stopped on its per-frame budget with more to
+ // remove. Leaving the rest to the next interval tick
+ // caps reclaim at one budget per interval, which a
+ // producer outruns forever on small segments. Re-stage
+ // as a frame rather than looping, so the pump keeps
+ // interleaving consensus ticks between passes. This
+ // terminates: a pass re-arms only when MORE than a full
+ // budget was removable and it retires exactly the
+ // budget, so the backlog strictly shrinks; a pass held
+ // by the consumer barrier stops at or below the budget
+ // and does not re-arm at all.
+ self.request_clean_partition(namespace, now, message_expiry, max_bytes);
+ }
}
}
LifecycleFrame::TruncatePartition {
@@ -575,9 +589,8 @@ where
// committed offset is identical on every replica, so the local
// deletion converges; idempotent if already trimmed past it.
if let Some(partition) = self.plane.partitions().get_mut_by_ns(&namespace) {
- let (segments, messages) =
- partition.remove_sealed_segments_up_to(up_to_offset).await;
- if segments > 0 {
+ let removal = partition.remove_sealed_segments_up_to(up_to_offset).await;
+ if removal.segments > 0 {
// See the cleaner arm: a truncate commits on the
// METADATA plane, so this partition's commit_op never
// moves and the cached offer stays a hit over unlinked
@@ -586,8 +599,8 @@ where
tracing::debug!(
shard = self.id,
namespace_raw = namespace.inner(),
- segments,
- messages,
+ segments = removal.segments,
+ messages = removal.messages,
up_to_offset,
"truncate-partition removed sealed segments"
);
diff --git a/examples/python/uv.lock b/examples/python/uv.lock
index bd2f00f01..137bd9a86 100644
--- a/examples/python/uv.lock
+++ b/examples/python/uv.lock
@@ -8,7 +8,7 @@ exclude-newer-span = "P7D"
[[package]]
name = "apache-iggy"
-version = "0.9.0.dev2"
+version = "0.9.0.dev3"
source = { directory = "../../foreign/python" }
[package.metadata]
diff --git a/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj b/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj
index 4708ee716..c0fe627d2 100644
--- a/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj
+++ b/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj
@@ -27,7 +27,7 @@
<TargetFrameworks>net8.0;net10.0</TargetFrameworks>
<AssemblyName>Apache.Iggy</AssemblyName>
<RootNamespace>Apache.Iggy</RootNamespace>
- <Version>0.9.0-edge.2</Version>
+ <Version>0.9.0-edge.3</Version>
<GenerateDocumentationFile>true</GenerateDocumentationFile>
</PropertyGroup>
diff --git a/foreign/go/contracts/version.go b/foreign/go/contracts/version.go
index 03840a8b2..ba3f36b4d 100644
--- a/foreign/go/contracts/version.go
+++ b/foreign/go/contracts/version.go
@@ -17,4 +17,4 @@
package iggcon
-const Version = "0.9.0-edge.2"
+const Version = "0.9.0-edge.3"
diff --git a/foreign/node/README.md b/foreign/node/README.md
index 994ffee60..e6f01a004 100644
--- a/foreign/node/README.md
+++ b/foreign/node/README.md
@@ -71,7 +71,13 @@ new session. Transient not-committed responses retry the exact encoded request
within one bounded deadline. A disconnected mutation is never replayed under a
new session.
-When the server's `[heartbeat]` eviction is enabled, configure the client's `heartbeatInterval` below the server heartbeat interval. Client heartbeats are disabled when `heartbeatInterval` is unset.
+The client pings every `heartbeatInterval` milliseconds, 5000 by default, which
+keeps an idle session alive when the server's `[heartbeat]` eviction is enabled.
+The server evicts a connection silent for 36 s, which is 1.2 x its 30 s
+heartbeat interval. Raising the client interval past that window, or setting it
+to 0 to disable client heartbeats, exposes an idle consumer-group member to
+eviction; a connection holding no group membership is left alone. Any other
+unusable value is rejected instead of silently disabling the heartbeat.
`sendBinaryRequest(code, payload)` sends an arbitrary command code. Known replicated commands use their registered operation, while unknown codes reach the server as non-replicated requests and are rejected by servers that do not register them.
diff --git a/foreign/node/package-lock.json b/foreign/node/package-lock.json
index 517e63988..e58a7d55b 100644
--- a/foreign/node/package-lock.json
+++ b/foreign/node/package-lock.json
@@ -1,12 +1,12 @@
{
"name": "apache-iggy",
- "version": "0.10.0-edge.2",
+ "version": "0.10.0-edge.3",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "apache-iggy",
- "version": "0.10.0-edge.2",
+ "version": "0.10.0-edge.3",
"license": "Apache-2.0",
"dependencies": {
"debug": "4.4.3",
diff --git a/foreign/node/package.json b/foreign/node/package.json
index 5696c58f0..c906d5cae 100644
--- a/foreign/node/package.json
+++ b/foreign/node/package.json
@@ -1,7 +1,7 @@
{
"name": "apache-iggy",
"type": "module",
- "version": "0.10.0-edge.2",
+ "version": "0.10.0-edge.3",
"description": "Official Apache Iggy NodeJS SDK",
"keywords": [
"iggy",
diff --git a/foreign/node/src/client/client.config.test.ts b/foreign/node/src/client/client.config.test.ts
index e4402e1b9..9022ab080 100644
--- a/foreign/node/src/client/client.config.test.ts
+++ b/foreign/node/src/client/client.config.test.ts
@@ -19,6 +19,7 @@ import assert from 'node:assert/strict';
import { describe, it } from 'node:test';
import type { ClientConfig } from './client.type.js';
import {
+ DEFAULT_HEARTBEAT_INTERVAL,
DEFAULT_MAX_RESPONSE_FRAME_SIZE,
normalizeClientConfig
} from './client.config.js';
@@ -39,6 +40,40 @@ describe('normalizeClientConfig', () => {
);
});
+ it('enables the heartbeat by default and honours an explicit interval', () => {
+ assert.equal(DEFAULT_HEARTBEAT_INTERVAL, 5000);
+
+ assert.equal(
+ normalizeClientConfig(config()).heartbeatInterval,
+ DEFAULT_HEARTBEAT_INTERVAL
+ );
+
+ assert.equal(
+ normalizeClientConfig({ ...config(), heartbeatInterval: 1000 })
+ .heartbeatInterval,
+ 1000
+ );
+
+ assert.equal(
+ normalizeClientConfig({ ...config(), heartbeatInterval: 0 })
+ .heartbeatInterval,
+ 0
+ );
+ });
+
+ it('rejects unusable heartbeat intervals', () => {
+ for (const heartbeatInterval of [
+ -1, -5000, Number.NaN, 1.5, 2_147_483_648, 2 ** 32, Number.MAX_VALUE
+ ])
+ assert.throws(
+ () => normalizeClientConfig({
+ ...config(),
+ heartbeatInterval
+ }),
+ /heartbeatInterval/
+ );
+ });
+
it('restricts the client to one pooled connection', () => {
const normalized = normalizeClientConfig(config());
assert.deepEqual(normalized.poolSize, { min: 1, max: 1 });
diff --git a/foreign/node/src/client/client.config.ts b/foreign/node/src/client/client.config.ts
index ed4030819..7c94bdbe4 100644
--- a/foreign/node/src/client/client.config.ts
+++ b/foreign/node/src/client/client.config.ts
@@ -19,6 +19,15 @@ import type { ClientConfig } from './client.type.js';
export const DEFAULT_MAX_RESPONSE_FRAME_SIZE = 64 * 1024 * 1024;
+/**
+ * The server evicts a connection silent for 36 s (1.2 x its 30 s heartbeat
+ * interval). 5 s matches the other SDKs and survives several skipped pings.
+ */
+export const DEFAULT_HEARTBEAT_INTERVAL = 5 * 1000;
+
+/** Node's largest timer delay: setInterval clamps anything above it to 1 ms. */
+export const MAX_HEARTBEAT_INTERVAL = 2_147_483_647;
+
export const normalizeClientConfig = (
config: ClientConfig
): ClientConfig => {
@@ -35,10 +44,24 @@ export const normalizeClientConfig = (
'VSR clients currently support exactly one pooled connection'
);
+ // Only 0 disables the heartbeat. Anything else unusable has to throw here:
+ // the default below is nullish-only, so a value that is negative, fractional
+ // or above MAX_HEARTBEAT_INTERVAL reaches setInterval, which clamps it to
+ // 1 ms and floods the server.
+ const heartbeatInterval =
+ config.heartbeatInterval ?? DEFAULT_HEARTBEAT_INTERVAL;
+ if (!Number.isSafeInteger(heartbeatInterval) ||
+ heartbeatInterval < 0 ||
+ heartbeatInterval > MAX_HEARTBEAT_INTERVAL)
+ throw new TypeError(
+ `heartbeatInterval must be a safe integer of milliseconds between 0 and ${MAX_HEARTBEAT_INTERVAL} (0 disables heartbeats)`
+ );
+
return {
...config,
options: { ...config.options },
maxResponseFrameSize,
+ heartbeatInterval,
poolSize: { min: 1, max: 1 }
};
};
diff --git a/foreign/node/src/client/client.socket.ts b/foreign/node/src/client/client.socket.ts
index a3aabcff5..d1e5e4fe2 100644
--- a/foreign/node/src/client/client.socket.ts
+++ b/foreign/node/src/client/client.socket.ts
@@ -625,6 +625,9 @@ export class CommandResponseStream extends EventEmitter {
}
}
}, interval);
+ // A pending heartbeat must not be the reason the process stays up: a script
+ // that never calls destroy() would otherwise hang on exit.
+ this.heartbeatIntervalHandler.unref();
}
/**
diff --git a/foreign/node/src/client/client.type.ts b/foreign/node/src/client/client.type.ts
index a7e63723a..4df9813e7 100644
--- a/foreign/node/src/client/client.type.ts
+++ b/foreign/node/src/client/client.type.ts
@@ -160,7 +160,12 @@ export type ClientConfig = {
poolSize?: PoolSizeOption,
/** Automatic reconnection configuration */
reconnect?: ReconnectOption,
- /** Interval for sending heartbeat pings in milliseconds */
+ /**
+ * Interval for sending heartbeat pings in milliseconds, as an integer
+ * between 0 and Node's timer ceiling. Defaults to 5000. Set to 0 to disable
+ * client heartbeats; any other unusable value is rejected rather than
+ * silently disabling them.
+ */
heartbeatInterval?: number,
/** Maximum accepted response frame size in bytes */
maxResponseFrameSize?: number
diff --git a/foreign/node/src/debug.ts b/foreign/node/src/debug.ts
index 2815ea04b..514951f35 100644
--- a/foreign/node/src/debug.ts
+++ b/foreign/node/src/debug.ts
@@ -17,7 +17,7 @@
//
import assert from 'node:assert/strict';
-import { Client } from './client/index.js';
+import { Client, DEFAULT_HEARTBEAT_INTERVAL } from './client/index.js';
import { uuidv7, uuidv4 } from 'uuidv7'
import { groupConsumerStream } from './stream/consumer-stream.js';
import { PollingStrategy, type PollMessagesResponse } from './wire/index.js';
@@ -46,7 +46,7 @@ const opt = {
interval: 10 * 1000,
maxRetries: 10
},
- heartbeatInterval: 5 * 1000
+ heartbeatInterval: DEFAULT_HEARTBEAT_INTERVAL
};
const c = new Client(opt);
diff --git a/foreign/python/Cargo.toml b/foreign/python/Cargo.toml
index 56f8891a9..cb6b3288d 100644
--- a/foreign/python/Cargo.toml
+++ b/foreign/python/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "apache-iggy"
-version = "0.9.0-dev2"
+version = "0.9.0-dev3"
edition = "2024"
authors = ["Iggy Committers <de...@iggy.apache.org>"]
license = "Apache-2.0"
diff --git a/foreign/python/pyproject.toml b/foreign/python/pyproject.toml
index 65332fce1..7f296302c 100644
--- a/foreign/python/pyproject.toml
+++ b/foreign/python/pyproject.toml
@@ -22,7 +22,7 @@ build-backend = "maturin"
[project]
name = "apache-iggy"
requires-python = ">=3.10"
-version = "0.9.0.dev2"
+version = "0.9.0.dev3"
description = "Apache Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
readme = "README.md"
license = { file = "LICENSE" }
diff --git a/foreign/python/uv.lock b/foreign/python/uv.lock
index a60f46845..d263c59cb 100644
--- a/foreign/python/uv.lock
+++ b/foreign/python/uv.lock
@@ -12,7 +12,7 @@ exclude-newer-span = "P7D"
[[package]]
name = "apache-iggy"
-version = "0.9.0.dev2"
+version = "0.9.0.dev3"
source = { editable = "." }
[package.optional-dependencies]
diff --git a/helm/charts/iggy/README.md b/helm/charts/iggy/README.md
index 4aba962b1..4e4cd4211 100644
--- a/helm/charts/iggy/README.md
+++ b/helm/charts/iggy/README.md
@@ -312,7 +312,7 @@ pre-commit install
| podSecurityContext | object | `{"seccompProfile":{"type":"Unconfined"}}` | Pod security context (server uses io_uring, requires unconfined seccomp) |
| resources | object | `{}` | Resource limits and requests for server |
| securityContext | object | `{"capabilities":{"add":["IPC_LOCK"]}}` | Container security context (server requires IPC_LOCK for io_uring) |
-| server | object | `{"affinity":{},"enabled":true,"env":[{"name":"RUST_LOG","value":"info"},{"name":"IGGY_HTTP_ADDRESS","value":"0.0.0.0:3000"},{"name":"IGGY_TCP_ADDRESS","value":"0.0.0.0:8090"},{"name":"IGGY_QUIC_ADDRESS","value":"0.0.0.0:8080"},{"name":"IGGY_WEBSOCKET_ADDRESS","value":"0.0.0.0:8092"}],"image":{"pullPolicy":"Always","repository":"apache/iggy","tag":"0.7.0"},"ingress":{"annotations":{},"className":"","enabled":false,"hosts":[{"host":"chart-example.local","paths":[{"path [...]
+| server | object | `{"affinity":{},"enabled":true,"env":[{"name":"RUST_LOG","value":"info"},{"name":"IGGY_HTTP_ADDRESS","value":"0.0.0.0:3000"},{"name":"IGGY_TCP_ADDRESS","value":"0.0.0.0:8090"},{"name":"IGGY_QUIC_ADDRESS","value":"0.0.0.0:8080"},{"name":"IGGY_WEBSOCKET_ADDRESS","value":"0.0.0.0:8092"}],"image":{"pullPolicy":"Always","repository":"apache/iggy","tag":"0.7.0"},"ingress":{"annotations":{},"className":"","enabled":false,"hosts":[{"host":"chart-example.local","paths":[{"path [...]
| server.affinity | object | `{}` | Affinity rules for server pods |
| server.enabled | bool | `true` | Enable the Iggy server deployment |
| server.env | list | `[{"name":"RUST_LOG","value":"info"},{"name":"IGGY_HTTP_ADDRESS","value":"0.0.0.0:3000"},{"name":"IGGY_TCP_ADDRESS","value":"0.0.0.0:8090"},{"name":"IGGY_QUIC_ADDRESS","value":"0.0.0.0:8080"},{"name":"IGGY_WEBSOCKET_ADDRESS","value":"0.0.0.0:8092"}]` | Environment variables for the server container |
@@ -338,6 +338,7 @@ pre-commit install
| server.service.port | int | `3000` | Service port for the server |
| server.service.type | string | `"ClusterIP"` | Service type for the server |
| server.serviceMonitor.additionalLabels | object | `{}` | Additional labels for the ServiceMonitor |
+| server.serviceMonitor.authorization | object | `{}` | Authorization for the scrape request. The metrics endpoint requires a bearer credential (e.g. a personal access token stored in a Secret): authorization: credentials: name: iggy-metrics-token key: token |
| server.serviceMonitor.enabled | bool | `false` | Enable ServiceMonitor for Prometheus Operator |
| server.serviceMonitor.honorLabels | bool | `false` | Honor labels from the target |
| server.serviceMonitor.interval | string | `"30s"` | Scrape interval (fallback to Prometheus default) |
diff --git a/helm/charts/iggy/templates/servicemonitor.yaml b/helm/charts/iggy/templates/servicemonitor.yaml
index ffff628b0..a50a5b152 100644
--- a/helm/charts/iggy/templates/servicemonitor.yaml
+++ b/helm/charts/iggy/templates/servicemonitor.yaml
@@ -31,6 +31,10 @@ spec:
endpoints:
- path: {{ .Values.server.serviceMonitor.path }}
port: http
+ {{- with .Values.server.serviceMonitor.authorization }}
+ authorization:
+ {{- toYaml . | nindent 6 }}
+ {{- end }}
namespaceSelector:
matchNames:
- {{ .Release.Namespace }}
diff --git a/helm/charts/iggy/values.yaml b/helm/charts/iggy/values.yaml
index 0c8e019f5..7e760ea54 100644
--- a/helm/charts/iggy/values.yaml
+++ b/helm/charts/iggy/values.yaml
@@ -57,6 +57,13 @@ server:
path: /metrics
# -- Timeout for scrape metrics request
scrapeTimeout: 10s
+ # -- Authorization for the scrape request. The metrics endpoint requires a
+ # bearer credential (e.g. a personal access token stored in a Secret):
+ # authorization:
+ # credentials:
+ # name: iggy-metrics-token
+ # key: token
+ authorization: {}
users:
root:
diff --git a/web/package-lock.json b/web/package-lock.json
index b8a2dc487..e356da2fe 100644
--- a/web/package-lock.json
+++ b/web/package-lock.json
@@ -1,12 +1,12 @@
{
"name": "iggy-web-ui",
- "version": "0.4.0-edge.1",
+ "version": "0.4.0-edge.2",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "iggy-web-ui",
- "version": "0.4.0-edge.1",
+ "version": "0.4.0-edge.2",
"dependencies": {
"@floating-ui/dom": "^1.8.0",
"@tailwindcss/postcss": "^4.3.3",
diff --git a/web/package.json b/web/package.json
index 8f66d85ea..eb2e63f03 100644
--- a/web/package.json
+++ b/web/package.json
@@ -1,6 +1,6 @@
{
"name": "iggy-web-ui",
- "version": "0.4.0-edge.1",
+ "version": "0.4.0-edge.2",
"private": true,
"scripts": {
"dev": "vite dev --port 3050",