This is an automated email from the ASF dual-hosted git repository. spetz pushed a commit to branch offset_hardening in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 7bd65595843b59e15d957141743c1844dab89840 Author: spetz <[email protected]> AuthorDate: Sat Sep 5 08:27:00 2026 +0200 fix(partitions): bound consumer offset state per partition --- Cargo.lock | 22 +- 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 +- core/binary_protocol/Cargo.toml | 2 +- core/binary_protocol/src/primitives/ack_level.rs | 6 +- .../consumer_offsets/delete_consumer_offset.rs | 4 +- .../consumer_offsets/store_consumer_offset.rs | 4 +- core/cli/Cargo.toml | 2 +- core/common/Cargo.toml | 2 +- core/common/src/error/iggy_error.rs | 2 +- core/common/src/traits/message_client.rs | 4 + core/connectors/sdk/Cargo.toml | 2 +- .../tests/cluster/consumer_offset_quota.rs | 42 +- .../tests/server/consumer_offset_quota_vsr.rs | 87 +++- core/message_bus/src/lib.rs | 2 +- core/metadata/src/impls/metadata.rs | 6 +- core/metadata/src/stm/user.rs | 4 +- core/partitions/src/consumer_offset_capacity.rs | 268 ++++++------- core/partitions/src/iggy_partition.rs | 445 ++++++++++++++++++--- core/partitions/src/iggy_partitions.rs | 16 + core/partitions/src/poll_plan.rs | 142 +++---- core/partitions/src/state_transfer.rs | 17 +- core/sdk/Cargo.toml | 2 +- core/sdk/src/clients/consumer.rs | 5 + core/server/Cargo.toml | 2 +- core/server/config.toml | 3 + core/server/src/boot/recovery.rs | 2 +- core/server/src/consumer_group.rs | 14 +- core/server/src/dispatch/partition.rs | 47 +-- core/server/src/partition_helpers.rs | 10 +- core/server/src/partition_reconciler.rs | 154 +++++-- core/server/src/responses.rs | 32 +- core/shard/src/lib.rs | 43 +- core/shard/src/metrics.rs | 21 +- core/shard/src/router.rs | 9 + core/simulator/src/lib.rs | 17 +- core/simulator/src/workload/effect.rs | 2 +- core/simulator/src/workload/shadow.rs | 2 +- examples/node/package-lock.json | 2 +- examples/python/uv.lock | 2 +- foreign/go/contracts/version.go | 2 +- foreign/go/errors/errors.yaml | 2 +- foreign/go/errors/errors_gen.go | 2 +- .../org/apache/iggy/exception/IggyErrorCode.java | 2 + foreign/node/package-lock.json | 4 +- foreign/node/package.json | 2 +- foreign/node/src/wire/error.code.test.ts | 2 +- foreign/node/src/wire/error.code.ts | 2 +- foreign/python/Cargo.toml | 4 +- foreign/python/pyproject.toml | 2 +- foreign/python/uv.lock | 2 +- 56 files changed, 1021 insertions(+), 475 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 56ecb71dd..7856dd413 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1862,7 +1862,7 @@ checksum = "3a8241f3ebb85c056b509d4327ad0358fbbba6ffb340bf388f26350aeda225b1" [[package]] name = "bench-dashboard-frontend" -version = "0.8.0-edge.3" +version = "0.8.0-edge.4" dependencies = [ "bench-dashboard-shared", "bench-report", @@ -1892,7 +1892,7 @@ dependencies = [ [[package]] name = "bench-report" -version = "0.4.0-edge.3" +version = "0.4.0-edge.4" dependencies = [ "charming", "colored", @@ -6594,7 +6594,7 @@ checksum = "cd62e6b5e86ea8eeeb8db1de02880a6abc01a397b2ebb64b5d74ac255318f5cb" [[package]] name = "iggy" -version = "0.11.0-edge.6" +version = "0.11.0-edge.7" dependencies = [ "async-broadcast", "async-dropper", @@ -6628,7 +6628,7 @@ dependencies = [ [[package]] name = "iggy-bench" -version = "0.6.0-edge.6" +version = "0.6.0-edge.7" dependencies = [ "async-trait", "bench-report", @@ -6657,7 +6657,7 @@ dependencies = [ [[package]] name = "iggy-bench-dashboard-server" -version = "0.8.0-edge.3" +version = "0.8.0-edge.4" dependencies = [ "actix-cors", "actix-files", @@ -6685,7 +6685,7 @@ dependencies = [ [[package]] name = "iggy-cli" -version = "0.14.0-edge.6" +version = "0.14.0-edge.7" dependencies = [ "anyhow", "apple-native-keyring-store", @@ -6791,7 +6791,7 @@ dependencies = [ [[package]] name = "iggy-mcp" -version = "0.5.0-edge.5" +version = "0.5.0-edge.6" dependencies = [ "axum", "axum-server", @@ -6825,7 +6825,7 @@ dependencies = [ [[package]] name = "iggy_binary_protocol" -version = "0.11.0-edge.6" +version = "0.11.0-edge.7" dependencies = [ "aligned-vec", "bytemuck", @@ -6838,7 +6838,7 @@ dependencies = [ [[package]] name = "iggy_common" -version = "0.11.0-edge.6" +version = "0.11.0-edge.7" dependencies = [ "aes-gcm 0.11.1", "async-broadcast", @@ -7203,7 +7203,7 @@ dependencies = [ [[package]] name = "iggy_connector_sdk" -version = "0.4.0-edge.3" +version = "0.4.0-edge.4" dependencies = [ "anyhow", "apache-avro 0.22.0", @@ -12160,7 +12160,7 @@ dependencies = [ [[package]] name = "server" -version = "0.9.0-edge.6" +version = "0.9.0-edge.7" dependencies = [ "ahash 0.8.12", "argon2", diff --git a/Cargo.toml b/Cargo.toml index af7a697ca..beb92e80f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -198,13 +198,13 @@ hyper-util = { version = "0.1.20", features = ["server-auto", "service"] } iceberg = "0.10.1" iceberg-catalog-rest = "0.10.1" iceberg-storage-opendal = "0.10.1" -iggy = { path = "core/sdk", version = "0.11.0-edge.6" } -iggy-cli = { path = "core/cli", version = "0.14.0-edge.6" } +iggy = { path = "core/sdk", version = "0.11.0-edge.7" } +iggy-cli = { path = "core/cli", version = "0.14.0-edge.7" } iggy-gateway-kafka = { path = "gateways/kafka" } -iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.0-edge.6" } -iggy_common = { path = "core/common", version = "0.11.0-edge.6" } +iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.0-edge.7" } +iggy_common = { path = "core/common", version = "0.11.0-edge.7" } iggy_connector_doris_sink = { path = "core/connectors/sinks/doris_sink" } -iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.4.0-edge.3" } +iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.4.0-edge.4" } indexmap = "2.14.1" integration = { path = "core/integration" } ipnet = "2.12.1" diff --git a/bdd/python/uv.lock b/bdd/python/uv.lock index 44c64ab55..b10e81b84 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.dev6" +version = "0.9.0.dev7" source = { directory = "../../foreign/python" } [package.metadata] diff --git a/core/ai/mcp/Cargo.toml b/core/ai/mcp/Cargo.toml index 5386c5c4a..1d1953f9f 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.5" +version = "0.5.0-edge.6" 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 e2a3c8d8d..7a3f1ce42 100644 --- a/core/bench/Cargo.toml +++ b/core/bench/Cargo.toml @@ -17,7 +17,7 @@ [package] name = "iggy-bench" -version = "0.6.0-edge.6" +version = "0.6.0-edge.7" 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 595cec1d0..6b023d631 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.3" +version = "0.8.0-edge.4" edition = "2024" publish = false diff --git a/core/bench/dashboard/server/Cargo.toml b/core/bench/dashboard/server/Cargo.toml index 726df33be..68570b112 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.3" +version = "0.8.0-edge.4" edition = "2024" publish = false diff --git a/core/bench/report/Cargo.toml b/core/bench/report/Cargo.toml index a418cdd25..bc5649a57 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.3" +version = "0.4.0-edge.4" 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/binary_protocol/Cargo.toml b/core/binary_protocol/Cargo.toml index 0a83a4c01..be97141e3 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.6" +version = "0.11.0-edge.7" 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/binary_protocol/src/primitives/ack_level.rs b/core/binary_protocol/src/primitives/ack_level.rs index a2ddb2389..a750b57d9 100644 --- a/core/binary_protocol/src/primitives/ack_level.rs +++ b/core/binary_protocol/src/primitives/ack_level.rs @@ -20,9 +20,9 @@ use crate::WireError; /// Acknowledgement policy for consumer-offset write commands. /// /// Wire format: single `u8` discriminant. -/// - `NoAck(0)`: leader-local write only; respond as soon as the in-memory -/// and on-disk state have been updated. Matches the fast path used by -/// `PollMessages` auto-commit. +/// - `NoAck(0)`: local fast path for a single-replica partition. Replicated +/// partitions commit offset writes through VSR before replying, including +/// when this acknowledgement value is selected. /// - `Quorum(1)`: submit through the partition VSR consensus pipeline and /// respond only after the write has been committed by a quorum of replicas. /// This is the default for explicit client writes. diff --git a/core/binary_protocol/src/requests/consumer_offsets/delete_consumer_offset.rs b/core/binary_protocol/src/requests/consumer_offsets/delete_consumer_offset.rs index 014081794..e5ca29d1a 100644 --- a/core/binary_protocol/src/requests/consumer_offsets/delete_consumer_offset.rs +++ b/core/binary_protocol/src/requests/consumer_offsets/delete_consumer_offset.rs @@ -24,8 +24,8 @@ use bytes::{BufMut, BytesMut}; /// `DeleteConsumerOffset` request. /// -/// Adds an `ack` byte: `NoAck` = leader-local fast path, `Quorum` = VSR -/// pipeline. +/// The `ack` byte selects the local fast path only for `NoAck` on a +/// single-replica partition. Replicated partitions use VSR for both values. /// /// Wire format: /// ```text diff --git a/core/binary_protocol/src/requests/consumer_offsets/store_consumer_offset.rs b/core/binary_protocol/src/requests/consumer_offsets/store_consumer_offset.rs index 5261d5c53..228360798 100644 --- a/core/binary_protocol/src/requests/consumer_offsets/store_consumer_offset.rs +++ b/core/binary_protocol/src/requests/consumer_offsets/store_consumer_offset.rs @@ -24,8 +24,8 @@ use bytes::{BufMut, BytesMut}; /// `StoreConsumerOffset` request. /// -/// Adds an `ack` byte: `NoAck` = leader-local fast path, `Quorum` = VSR -/// pipeline. +/// The `ack` byte selects the local fast path only for `NoAck` on a +/// single-replica partition. Replicated partitions use VSR for both values. /// /// Wire format: /// ```text diff --git a/core/cli/Cargo.toml b/core/cli/Cargo.toml index a2f8756bb..f8a2b89f1 100644 --- a/core/cli/Cargo.toml +++ b/core/cli/Cargo.toml @@ -17,7 +17,7 @@ [package] name = "iggy-cli" -version = "0.14.0-edge.6" +version = "0.14.0-edge.7" edition = "2024" rust-version.workspace = true authors = ["[email protected]"] diff --git a/core/common/Cargo.toml b/core/common/Cargo.toml index 16aaa621c..2e059fc3e 100644 --- a/core/common/Cargo.toml +++ b/core/common/Cargo.toml @@ -17,7 +17,7 @@ [package] name = "iggy_common" -version = "0.11.0-edge.6" +version = "0.11.0-edge.7" 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/error/iggy_error.rs b/core/common/src/error/iggy_error.rs index b312d61dc..e77abebc5 100644 --- a/core/common/src/error/iggy_error.rs +++ b/core/common/src/error/iggy_error.rs @@ -333,7 +333,7 @@ pub enum IggyError { NotResolvedConsumer(Identifier) = 3022, #[error("Cannot open consumer offsets file for path: {0}")] CannotOpenConsumerOffsetsFile(String) = 3023, - #[error("Too many consumer offsets")] + #[error("Too many consumer offsets for partition")] TooManyConsumerOffsets = 3024, #[error("Segment not found")] SegmentNotFound = 4000, diff --git a/core/common/src/traits/message_client.rs b/core/common/src/traits/message_client.rs index 18c6d2294..7e1dcd026 100644 --- a/core/common/src/traits/message_client.rs +++ b/core/common/src/traits/message_client.rs @@ -31,6 +31,10 @@ pub trait MessageClient { /// Polling a consumer group the client is not (or no longer) a member of fails with `ConsumerGroupMemberNotFound` rather than returning an empty batch, so the caller can rejoin. /// A member that holds no partitions gets an empty batch whose `partition_id` is [`NO_ASSIGNED_PARTITION`](crate::NO_ASSIGNED_PARTITION). #[allow(clippy::too_many_arguments)] + /// With server-side auto-commit enabled, a new consumer offset key can be + /// rejected with `TooManyConsumerOffsets` at the partition's configured + /// limit. That poll returns no messages. Existing keys remain writable, + /// and polling without auto-commit does not allocate a stored offset. async fn poll_messages( &self, stream_id: &Identifier, diff --git a/core/connectors/sdk/Cargo.toml b/core/connectors/sdk/Cargo.toml index fae1b4a0a..5d6b98daa 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.3" +version = "0.4.0-edge.4" 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/integration/tests/cluster/consumer_offset_quota.rs b/core/integration/tests/cluster/consumer_offset_quota.rs index c39bab514..3aa9a7d9d 100644 --- a/core/integration/tests/cluster/consumer_offset_quota.rs +++ b/core/integration/tests/cluster/consumer_offset_quota.rs @@ -141,6 +141,22 @@ async fn given_replicated_partition_when_no_ack_offsets_mutate_should_converge_a ) .await; + let auto_commit_consumer = Consumer::new(Identifier::numeric(1).expect("consumer identifier")); + let polled = client + .poll_messages( + &stream, + &topic, + Some(PARTITION_ID), + &auto_commit_consumer, + &PollingStrategy::first(), + 1, + true, + ) + .await + .expect("auto-commit poll within limit"); + assert_eq!(polled.messages.len(), 1); + wait_for_file_state(harness, stream_details.id, topic_details.id, 1, true).await; + for consumer_id in 1..=4 { client .store_consumer_offset( @@ -306,23 +322,15 @@ async fn wait_for_max_file_count(harness: &TestHarness, stream_id: u32, topic_id loop { let counts: Vec<usize> = (0..harness.cluster_size()) .map(|node| { - let dir = offset_file(harness, node, stream_id, topic_id, 0) - .parent() - .expect("offset file has parent") - .to_path_buf(); - std::fs::read_dir(dir) - .map(|entries| { - entries - .filter_map(Result::ok) - .filter(|entry| { - entry - .file_name() - .to_str() - .is_some_and(|name| name.parse::<u32>().is_ok()) - }) - .count() - }) - .unwrap_or_default() + disk::consumer_offset_file_ids( + &harness.node(node).data_path(), + stream_id, + topic_id, + PARTITION_ID, + ConsumerKind::Consumer, + ) + .expect("consumer offset directory") + .len() }) .collect(); if counts.iter().all(|count| *count == max) { diff --git a/core/integration/tests/server/consumer_offset_quota_vsr.rs b/core/integration/tests/server/consumer_offset_quota_vsr.rs index 3849ae096..2ccac51c4 100644 --- a/core/integration/tests/server/consumer_offset_quota_vsr.rs +++ b/core/integration/tests/server/consumer_offset_quota_vsr.rs @@ -107,6 +107,49 @@ async fn given_full_consumer_offset_table_when_creating_another_should_reject_wi .await .expect("consumer login"); + let first_consumer = Consumer::new(Identifier::numeric(1).unwrap()); + let polled = client + .poll_messages( + &stream, + &topic, + Some(PARTITION_ID), + &first_consumer, + &PollingStrategy::first(), + 1, + true, + ) + .await + .expect("new auto-commit consumer fits"); + assert_eq!(polled.messages.len(), 1); + let first_file = harness.server().data_path().join(format!( + "streams/{}/topics/{}/partitions/{PARTITION_ID}/offsets/consumers/1", + stream_details.id, topic_details.id + )); + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10); + while !first_file.is_file() { + assert!( + tokio::time::Instant::now() < deadline, + "auto-commit never reached its file" + ); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + assert!( + client + .poll_messages( + &stream, + &topic, + Some(PARTITION_ID), + &first_consumer, + &PollingStrategy::next(), + 1, + true + ) + .await + .expect("next poll") + .messages + .is_empty() + ); + for consumer_id in 1..=LIMIT { client .store_consumer_offset( @@ -205,23 +248,23 @@ async fn given_full_consumer_offset_table_when_creating_another_should_reject_wi let (reply, _) = raw_tcp::exchange(&mut raw, &header, &unresolved_group).await; assert_eq!( raw_tcp::reply_status(&reply), - IggyError::InvalidIdentifier.as_code() + IggyError::ConsumerGroupIdNotFound(Identifier::numeric(999).unwrap(), topic.clone()) + .as_code() ); let offsets_dir = harness.server().data_path().join(format!( "streams/{}/topics/{}/partitions/{PARTITION_ID}/offsets/consumers", stream_details.id, topic_details.id )); - let file_count = fs::read_dir(&offsets_dir) - .expect("consumer offsets directory") - .filter_map(Result::ok) - .filter(|entry| { - entry - .file_name() - .to_str() - .is_some_and(|name| name.parse::<u32>().is_ok()) - }) - .count(); + let file_count = integration::harness::disk::consumer_offset_file_ids( + &harness.server().data_path(), + stream_details.id, + topic_details.id, + PARTITION_ID, + ConsumerKind::Consumer, + ) + .expect("consumer offsets directory") + .len(); assert_eq!(file_count, LIMIT as usize); let groups_dir = offsets_dir .parent() @@ -232,6 +275,28 @@ async fn given_full_consumer_offset_table_when_creating_another_should_reject_wi .unwrap_or_default(); assert_eq!(group_file_count, 0); + let named_group = StoreConsumerOffsetRequest { + consumer: WireConsumer::consumer_group(WireIdentifier::named("unknown-group").unwrap()), + stream_id: WireIdentifier::Numeric(stream_details.id), + topic_id: WireIdentifier::Numeric(topic_details.id), + partition_id: Some(PARTITION_ID), + offset: 0, + ack: AckLevel::Quorum, + } + .to_bytes(); + let header = raw_tcp::request_header( + Operation::StoreConsumerOffset, + raw_client_id, + session, + 2, + named_group.len(), + ); + let (reply, _) = raw_tcp::exchange(&mut raw, &header, &named_group).await; + assert_eq!( + raw_tcp::reply_status(&reply), + IggyError::ConsumerGroupNameNotFound("unknown-group".to_owned(), topic.clone()).as_code() + ); + let http = HttpClient::login_root(harness).await; let response = http .client diff --git a/core/message_bus/src/lib.rs b/core/message_bus/src/lib.rs index 74836bb8e..d8038b612 100644 --- a/core/message_bus/src/lib.rs +++ b/core/message_bus/src/lib.rs @@ -797,7 +797,7 @@ impl IggyMessageBus { } /// Construct a bus with explicit runtime tunables and a pre-allocated - /// owner table. Server-ng bootstrap uses this so every shard's bus + /// owner table. Server bootstrap uses this so every shard's bus /// shares the same atomic slots; tests use [`Self::with_tunables`] /// which allocates a fresh table per bus. /// diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index 726a013c1..5358fd0d4 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -679,7 +679,7 @@ pub fn apply_committed_prepare<M>( pub type CommitNotifier = std::rc::Rc<dyn Fn(Operation)>; pub struct IggyMetadata<C, J, S, M, SB = PingPongSuperblock> { - /// `Some` on shard 0, `None` on other shards. Server-ng bootstrap + /// `Some` on shard 0, `None` on other shards. Server bootstrap /// holds the invariant: only shard 0 owns the metadata consensus /// replica; every other shard reconstructs `mux_stm` from the /// `MetadataHandoff::Waiter` factory bundle broadcast by shard 0 @@ -897,7 +897,7 @@ impl<C, J, S, M, SB> IggyMetadata<C, J, S, M, SB> { } /// Install (or replace) the post-commit notifier. Passing `None` - /// removes any previous one. Server-ng bootstrap calls this on shard 0 + /// removes any previous one. Server bootstrap calls this on shard 0 /// only; peer shards never commit metadata locally. pub fn set_commit_notifier(&self, notifier: Option<CommitNotifier>) { *self.commit_notifier.borrow_mut() = notifier; @@ -906,7 +906,7 @@ impl<C, J, S, M, SB> IggyMetadata<C, J, S, M, SB> { /// Seed the coordinator's last-checkpoint pairing at boot from the recovered /// snapshot, so the first post-boot view-change superblock write records the real /// `(checkpoint_op, checksum)` instead of `(0, 0)`. No-op without a coordinator - /// (peer shards, the simulator). Server-ng bootstrap calls this on shard 0 after + /// (peer shards, the simulator). Server bootstrap calls this on shard 0 after /// cross-checking the pairing. pub fn seed_checkpoint_ref(&self, checkpoint_op: u64, checkpoint_checksum: u128) { if let Some(coordinator) = &self.coordinator { diff --git a/core/metadata/src/stm/user.rs b/core/metadata/src/stm/user.rs index 1e79d7853..c78ea7507 100644 --- a/core/metadata/src/stm/user.rs +++ b/core/metadata/src/stm/user.rs @@ -642,14 +642,14 @@ impl StateHandler for UpdatePermissionsRequest { /// The success reply here is deliberately empty: the raw token the caller needs /// is the one thing this apply must never see. /// -/// The primary mints the raw token and its hash at ingress (server-ng +/// The primary mints the raw token and its hash at ingress (server /// `pat::rewrite_pat_request_for_user`) and replicates only the hash. Minting /// inside this apply would call `ring::rand` on every replica and diverge the /// token index, and replicating the raw token would persist a live credential in /// every WAL and snapshot. So the raw token leaves the primary by a side channel /// (`maybe_rewrite_pat_request` returns it alongside the rewritten request) and /// the home shard splices it into this op's reply as a typed -/// `RawPersonalAccessTokenResponse` (server-ng `responses::build_raw_pat_reply`). +/// `RawPersonalAccessTokenResponse` (server `responses::build_raw_pat_reply`). /// /// One consequence rides on that: the secret exists only on the wire of the /// original reply, so a replayed request cannot be served from the client-table diff --git a/core/partitions/src/consumer_offset_capacity.rs b/core/partitions/src/consumer_offset_capacity.rs index 6523b2614..27c786ceb 100644 --- a/core/partitions/src/consumer_offset_capacity.rs +++ b/core/partitions/src/consumer_offset_capacity.rs @@ -19,6 +19,7 @@ use iggy_common::ConsumerKind; use std::cell::{Cell, RefCell}; use std::collections::{HashMap, HashSet}; use std::rc::Rc; +use std::sync::{Arc, Weak}; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct DurableOffsetState { @@ -28,25 +29,30 @@ pub struct DurableOffsetState { #[derive(Debug, Default)] pub struct DurableConsumerOffsets { - entries: RefCell<HashMap<(ConsumerKind, u32), DurableOffsetState>>, - consumer_count: Cell<usize>, - group_count: Cell<usize>, + consumers: RefCell<HashMap<u32, DurableOffsetState>>, + groups: RefCell<HashMap<u32, DurableOffsetState>>, } impl DurableConsumerOffsets { pub(crate) fn get(&self, kind: ConsumerKind, id: u32) -> Option<DurableOffsetState> { - self.entries.borrow().get(&(kind, id)).copied() + self.entries(kind).borrow().get(&id).copied() } pub(crate) fn contains(&self, kind: ConsumerKind, id: u32) -> bool { - self.entries.borrow().contains_key(&(kind, id)) + self.entries(kind).borrow().contains_key(&id) } - pub(crate) const fn count(&self, kind: ConsumerKind) -> usize { - match kind { - ConsumerKind::Consumer => self.consumer_count.get(), - ConsumerKind::ConsumerGroup => self.group_count.get(), - } + pub(crate) fn count(&self, kind: ConsumerKind) -> usize { + self.entries(kind).borrow().len() + } + + pub(crate) fn covers(&self, kind: ConsumerKind, id: u32, offset: u64) -> bool { + self.get(kind, id).is_some_and(|state| { + state.committed_offset >= offset + && state + .persisted_high_water + .is_some_and(|persisted| persisted >= offset) + }) } pub(crate) fn record_explicit( @@ -56,8 +62,7 @@ impl DurableConsumerOffsets { committed_offset: u64, persisted_high_water: Option<u64>, ) { - self.insert_or_update( - kind, + self.entries(kind).borrow_mut().insert( id, DurableOffsetState { committed_offset, @@ -71,99 +76,70 @@ impl DurableConsumerOffsets { kind: ConsumerKind, id: u32, committed_offset: u64, - persisted_high_water: Option<u64>, + persisted_high_water: u64, ) { - let mut entries = self.entries.borrow_mut(); - if let Some(state) = entries.get_mut(&(kind, id)) { - state.committed_offset = state.committed_offset.max(committed_offset); - if let Some(high_water) = persisted_high_water { - state.persisted_high_water = Some( - state - .persisted_high_water - .map_or(high_water, |current| current.max(high_water)), - ); - } - return; - } - entries.insert( - (kind, id), - DurableOffsetState { - committed_offset, - persisted_high_water, - }, + let mut entries = self.entries(kind).borrow_mut(); + let state = entries.entry(id).or_insert(DurableOffsetState { + committed_offset, + persisted_high_water: None, + }); + state.committed_offset = state.committed_offset.max(committed_offset); + state.persisted_high_water = Some( + state + .persisted_high_water + .unwrap_or(0) + .max(persisted_high_water), ); - drop(entries); - self.increment(kind); } pub(crate) fn mark_persisted(&self, kind: ConsumerKind, id: u32, high_water: u64) { - if let Some(state) = self.entries.borrow_mut().get_mut(&(kind, id)) { + if let Some(state) = self.entries(kind).borrow_mut().get_mut(&id) { state.persisted_high_water = Some(high_water); } } pub(crate) fn remove(&self, kind: ConsumerKind, id: u32) -> bool { - if self.entries.borrow_mut().remove(&(kind, id)).is_none() { - return false; - } - self.decrement(kind); - true + self.entries(kind).borrow_mut().remove(&id).is_some() } pub(crate) fn clear(&self) { - self.entries.borrow_mut().clear(); - self.consumer_count.set(0); - self.group_count.set(0); + self.consumers.borrow_mut().clear(); + self.groups.borrow_mut().clear(); } pub(crate) fn committed_entries(&self, kind: ConsumerKind) -> Vec<(u32, u64)> { - self.entries + self.entries(kind) .borrow() .iter() - .filter_map(|((entry_kind, id), state)| { - (*entry_kind == kind).then_some((*id, state.committed_offset)) - }) + .map(|(id, state)| (*id, state.committed_offset)) .collect() } - fn insert_or_update(&self, kind: ConsumerKind, id: u32, state: DurableOffsetState) { - let inserted = self - .entries - .borrow_mut() - .insert((kind, id), state) - .is_none(); - if inserted { - self.increment(kind); - } - } - - fn increment(&self, kind: ConsumerKind) { - match kind { - ConsumerKind::Consumer => self.consumer_count.set(self.consumer_count.get() + 1), - ConsumerKind::ConsumerGroup => self.group_count.set(self.group_count.get() + 1), - } - } - - fn decrement(&self, kind: ConsumerKind) { + const fn entries(&self, kind: ConsumerKind) -> &RefCell<HashMap<u32, DurableOffsetState>> { match kind { - ConsumerKind::Consumer => self.consumer_count.set(self.consumer_count.get() - 1), - ConsumerKind::ConsumerGroup => self.group_count.set(self.group_count.get() - 1), + ConsumerKind::Consumer => &self.consumers, + ConsumerKind::ConsumerGroup => &self.groups, } } } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum CapacityReservation { - Existing, - Reserved, -} - #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct ConsumerOffsetCapacityError { pub kind: ConsumerKind, pub occupied: usize, pub limit: usize, pub first_in_episode: bool, + pub uncertain: bool, +} + +impl From<ConsumerOffsetCapacityError> for iggy_common::IggyError { + fn from(error: ConsumerOffsetCapacityError) -> Self { + if error.uncertain { + Self::TransientNotAccepted + } else { + Self::TooManyConsumerOffsets + } + } } #[derive(Debug)] @@ -171,7 +147,7 @@ pub struct ConsumerOffsetCapacity { kind: ConsumerKind, limit: Cell<usize>, pending: RefCell<HashMap<u32, usize>>, - provisional: RefCell<HashMap<u32, usize>>, + provisional: RefCell<HashMap<u32, Weak<()>>>, stranded: RefCell<HashSet<u32>>, uncertain: Cell<bool>, durable_warned: Cell<bool>, @@ -196,43 +172,52 @@ impl ConsumerOffsetCapacity { self.limit.set(limit); } + pub(crate) const fn limit(&self) -> usize { + self.limit.get() + } + pub(crate) fn try_reserve( &self, id: u32, durable: &DurableConsumerOffsets, - ) -> Result<CapacityReservation, ConsumerOffsetCapacityError> { - let existing = self.check(id, durable)?; + ) -> Result<(), ConsumerOffsetCapacityError> { + self.check(id, durable)?; *self.pending.borrow_mut().entry(id).or_default() += 1; - Ok(if existing { - CapacityReservation::Existing - } else { - CapacityReservation::Reserved - }) + Ok(()) } pub(crate) fn check( &self, id: u32, durable: &DurableConsumerOffsets, - ) -> Result<bool, ConsumerOffsetCapacityError> { + ) -> Result<(), ConsumerOffsetCapacityError> { + self.rearm_if_below_limit(durable); if durable.contains(self.kind, id) || self.pending.borrow().contains_key(&id) - || self.provisional.borrow().contains_key(&id) + || self + .provisional + .borrow() + .get(&id) + .is_some_and(|token| token.strong_count() > 0) || self.stranded.borrow().contains(&id) { - return Ok(true); + return Ok(()); } + self.provisional + .borrow_mut() + .retain(|_, token| token.strong_count() > 0); let occupied = self.occupied(durable); let limit = self.limit.get(); if self.uncertain.get() || occupied >= limit { return Err(ConsumerOffsetCapacityError { kind: self.kind, - occupied: occupied.max(limit), + occupied, limit, first_in_episode: !self.durable_warned.replace(true), + uncertain: self.uncertain.get(), }); } - Ok(false) + Ok(()) } pub(crate) fn reserve_provisional( @@ -241,14 +226,41 @@ impl ConsumerOffsetCapacity { durable: &Rc<DurableConsumerOffsets>, ) -> Result<AutoCommitReservation, ConsumerOffsetCapacityError> { self.check(id, durable)?; - *self.provisional.borrow_mut().entry(id).or_default() += 1; + let mut provisional = self.provisional.borrow_mut(); + let token = provisional + .get(&id) + .and_then(Weak::upgrade) + .unwrap_or_else(|| { + let token = Arc::new(()); + provisional.insert(id, Arc::downgrade(&token)); + token + }); Ok(AutoCommitReservation { - capacity: Rc::clone(self), - durable: Rc::clone(durable), + token, + kind: self.kind, consumer_id: id, }) } + pub(crate) fn owns(&self, reservation: &AutoCommitReservation) -> bool { + reservation.kind == self.kind + && self + .provisional + .borrow() + .get(&reservation.consumer_id) + .is_some_and(|token| std::ptr::eq(token.as_ptr(), Arc::as_ptr(&reservation.token))) + } + + pub(crate) fn protects(&self, id: u32, durable: &DurableConsumerOffsets) -> bool { + durable.contains(self.kind, id) + || self.pending.borrow().contains_key(&id) + || self + .provisional + .borrow() + .get(&id) + .is_some_and(|token| token.strong_count() > 0) + } + pub(crate) fn set_pending_count(&self, id: u32, count: usize) { if count == 0 { self.pending.borrow_mut().remove(&id); @@ -257,11 +269,6 @@ impl ConsumerOffsetCapacity { } } - pub(crate) fn promote_to_durable(&self, id: u32) { - self.release_reservation(id); - self.stranded.borrow_mut().remove(&id); - } - pub(crate) fn release_reservation(&self, id: u32) { let mut pending = self.pending.borrow_mut(); let Some(count) = pending.get_mut(&id) else { @@ -315,6 +322,9 @@ impl ConsumerOffsetCapacity { } pub(crate) fn rearm_if_below_limit(&self, durable: &DurableConsumerOffsets) { + if !self.durable_warned.get() || self.uncertain.get() { + return; + } if self.occupied(durable) < self.limit.get() { self.durable_warned.set(false); } @@ -333,6 +343,7 @@ impl ConsumerOffsetCapacity { occupied: map_len, limit, first_in_episode: !self.map_warned.replace(true), + uncertain: false, }) } @@ -342,51 +353,36 @@ impl ConsumerOffsetCapacity { } } - pub(crate) fn occupied_count(&self, durable: &DurableConsumerOffsets) -> usize { - self.occupied(durable) - } - - fn occupied(&self, durable: &DurableConsumerOffsets) -> usize { + pub(crate) fn occupied(&self, durable: &DurableConsumerOffsets) -> usize { let pending = self.pending.borrow(); let provisional = self.provisional.borrow(); let stranded = self.stranded.borrow(); durable.count(self.kind) + pending .keys() - .chain(provisional.keys().filter(|id| !pending.contains_key(id))) .chain( - stranded + provisional .iter() - .filter(|id| !pending.contains_key(id) && !provisional.contains_key(id)), + .filter(|(id, token)| !pending.contains_key(id) && token.strong_count() > 0) + .map(|(id, _)| id), ) + .chain(stranded.iter().filter(|id| { + !pending.contains_key(id) + && provisional + .get(id) + .is_none_or(|token| token.strong_count() == 0) + })) .filter(|id| !durable.contains(self.kind, **id)) .count() } } -/// Owns a poll's reservation until its submitted operation either commits or -/// loses its reply channel. -/// -/// Journal reservations are accounted independently, -/// so cancellation and view changes cannot release another operation's slot. +/// Keeps a provisional key occupied until the pump admits or drops its request. +#[derive(Debug)] pub struct AutoCommitReservation { - capacity: Rc<ConsumerOffsetCapacity>, - durable: Rc<DurableConsumerOffsets>, - consumer_id: u32, -} - -impl Drop for AutoCommitReservation { - fn drop(&mut self) { - let mut provisional = self.capacity.provisional.borrow_mut(); - if let Some(count) = provisional.get_mut(&self.consumer_id) { - *count -= 1; - if *count == 0 { - provisional.remove(&self.consumer_id); - } - } - drop(provisional); - self.capacity.rearm_if_below_limit(&self.durable); - } + token: Arc<()>, + pub(crate) kind: ConsumerKind, + pub(crate) consumer_id: u32, } #[cfg(test)] @@ -421,9 +417,9 @@ mod tests { assert!(capacity.check(9, &durable).is_err()); drop(first); assert!(capacity.check(9, &durable).is_ok()); - assert_eq!(capacity.occupied_count(&durable), 1); + assert_eq!(capacity.occupied(&durable), 1); drop(second); - assert_eq!(capacity.occupied_count(&durable), 0); + assert_eq!(capacity.occupied(&durable), 0); } #[test] @@ -443,14 +439,8 @@ mod tests { fn given_same_pending_key_when_reserved_twice_should_consume_one_slot() { let durable = DurableConsumerOffsets::default(); let capacity = ConsumerOffsetCapacity::new(ConsumerKind::Consumer, 1); - assert_eq!( - capacity.try_reserve(7, &durable), - Ok(CapacityReservation::Reserved) - ); - assert_eq!( - capacity.try_reserve(7, &durable), - Ok(CapacityReservation::Existing) - ); + assert_eq!(capacity.try_reserve(7, &durable), Ok(())); + assert_eq!(capacity.try_reserve(7, &durable), Ok(())); assert!(capacity.try_reserve(8, &durable).is_err()); capacity.release_reservation(7); assert!( @@ -484,7 +474,15 @@ mod tests { let error = capacity .try_reserve(7, &durable) .expect_err("unknown pending state must block new keys"); - assert_eq!(error.occupied, 4); + assert_eq!(error.occupied, 0); + assert!(error.uncertain); + assert!(error.first_in_episode); + assert!( + !capacity + .try_reserve(8, &durable) + .expect_err("the same uncertain episode stays closed") + .first_in_episode + ); } #[test] diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index b42d8cb01..352dc60ea 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -16,8 +16,7 @@ // under the License. use crate::consumer_offset_capacity::{ - CapacityReservation, ConsumerOffsetCapacity, ConsumerOffsetCapacityError, - DurableConsumerOffsets, + ConsumerOffsetCapacity, ConsumerOffsetCapacityError, DurableConsumerOffsets, }; use crate::iggy_index_writer::IggyIndexWriter; use crate::journal::{MessageLookup, PartitionJournal, PartitionJournalMemStorage}; @@ -209,6 +208,8 @@ where /// server down without the partition moving again in the meantime. fatal: Option<FatalCommit>, pub(crate) pending_consumer_offset_commits: HashMap<u64, PendingConsumerOffsetCommit>, + pub(crate) queued_auto_commit_reservations: + RefCell<HashMap<(ConsumerKind, u32), Vec<crate::AutoCommitReservation>>>, /// Committed consumer-offset membership and values. This is deliberately /// separate from the eager poll maps because follower-local and uncommitted /// auto-commit progress must never consume a durable slot or enter a state @@ -556,6 +557,7 @@ where installed_frontier: None, fatal: None, pending_consumer_offset_commits: HashMap::new(), + queued_auto_commit_reservations: RefCell::new(HashMap::new()), durable_consumer_offsets: Rc::new(DurableConsumerOffsets::default()), consumer_offset_capacity: Rc::new(ConsumerOffsetCapacity::new( ConsumerKind::Consumer, @@ -2009,11 +2011,13 @@ where ); } - pub fn seed_stranded_consumer_offset(&self, kind: ConsumerKind, consumer_id: u32) { + pub fn seed_stranded_consumer_offset(&self, kind: ConsumerKind, consumer_id: u32) -> bool { if !self.durable_consumer_offsets.contains(kind, consumer_id) { self.consumer_offset_capacity_for(kind) .record_stranded(consumer_id); + return true; } + false } #[must_use] @@ -2024,7 +2028,7 @@ where #[must_use] pub fn occupied_consumer_offset_count(&self, kind: ConsumerKind) -> usize { self.consumer_offset_capacity_for(kind) - .occupied_count(&self.durable_consumer_offsets) + .occupied(&self.durable_consumer_offsets) } #[must_use] @@ -2118,6 +2122,10 @@ where ) -> Result<(), IggyError> { let path = self.persisted_offset_path(pending.kind, pending.consumer_id); let capacity = self.consumer_offset_capacity_for(pending.kind); + let creates_group = pending.kind == ConsumerKind::ConsumerGroup + && !self + .durable_consumer_offsets + .contains(pending.kind, pending.consumer_id); match pending.mutation { // A server auto-commit persists monotonically: its op offset can // trail the durably-recorded value (disk-tier polls replicate in @@ -2134,35 +2142,37 @@ where .durable_consumer_offsets .get(pending.kind, pending.consumer_id); let persisted_high_water = match (path.as_deref(), tracked) { - (None, _) => Some(offset), + (None, _) => offset, (Some(_), Some(state)) if state .persisted_high_water .is_some_and(|high_water| offset <= high_water) => { - state.persisted_high_water + state.persisted_high_water.expect("covered persisted value") } (Some(path), Some(state)) => { let value = state.committed_offset.max(offset); persist_offset(path, value, self.consumer_offset_enforce_fsync).await?; - Some(value) + value + } + (Some(path), None) => { + persist_offset_max(path, offset, self.consumer_offset_enforce_fsync).await? } - (Some(path), None) => Some( - persist_offset_max(path, offset, self.consumer_offset_enforce_fsync) - .await?, - ), }; self.durable_consumer_offsets.record_auto_commit( pending.kind, pending.consumer_id, if tracked.is_none() { - persisted_high_water.unwrap_or(offset) + persisted_high_water } else { offset }, persisted_high_water, ); - capacity.promote_to_durable(pending.consumer_id); + capacity.clear_stranded(pending.consumer_id); + if creates_group { + self.consumer_group_offsets_need_reconcile.set(true); + } Ok(()) } PendingConsumerOffsetMutation::Upsert(offset) => { @@ -2175,7 +2185,10 @@ where offset, Some(offset), ); - capacity.promote_to_durable(pending.consumer_id); + capacity.clear_stranded(pending.consumer_id); + if creates_group { + self.consumer_group_offsets_need_reconcile.set(true); + } Ok(()) } PendingConsumerOffsetMutation::Delete => { @@ -2206,13 +2219,7 @@ where offset: u64, ) -> bool { self.durable_consumer_offsets - .get(kind, consumer_id) - .is_some_and(|state| { - state.committed_offset >= offset - && state - .persisted_high_water - .is_some_and(|high_water| offset <= high_water) - }) + .covers(kind, consumer_id, offset) } /// Reject completion from a removed partition or a view whose retained @@ -2222,13 +2229,22 @@ where applied.belongs_to(&self.durable_consumer_offsets) && self.observed_view == self.consensus.view() && !self.offset_reservations_need_resync.get() - && !self + && (!self .consumer_offset_capacity_for(applied.kind) .is_uncertain() + || self + .durable_consumer_offsets + .contains(applied.kind, applied.consumer_id)) } fn apply_consumer_offset_commit(&self, pending: PendingConsumerOffsetCommit) { - if pending.kind == ConsumerKind::ConsumerGroup { + if pending.kind == ConsumerKind::ConsumerGroup + && (matches!(pending.mutation, PendingConsumerOffsetMutation::Delete) + || !self + .consumer_group_offsets + .pin() + .contains_key(&ConsumerGroupId(pending.consumer_id as usize))) + { self.consumer_group_offsets_need_reconcile.set(true); } match pending.mutation { @@ -2434,6 +2450,9 @@ where &request_header, committed_reply_body(request_header.operation), ); + if offset.is_some() { + self.release_consumer_offset_reservation(kind, consumer_id); + } // Same rule as the committed path: a submit's waiter takes the reply, // because `header.client` is then the VSR consensus id. if let Some(waiter) = waiter { @@ -2487,7 +2506,7 @@ where &self, kind: ConsumerKind, consumer_id: u32, - ) -> Result<CapacityReservation, ConsumerOffsetCapacityError> { + ) -> Result<(), ConsumerOffsetCapacityError> { self.consumer_offset_capacity_for(kind) .try_reserve(consumer_id, &self.durable_consumer_offsets) } @@ -2519,7 +2538,9 @@ where consumer_id: u32, waiter: &mut Option<consensus::Sender<Message<ReplyHeader>>>, ) -> bool { - if self.consumer_offset_capacity_for(kind).is_uncertain() { + if self.consumer_offset_capacity_for(kind).is_uncertain() + && !self.durable_consumer_offsets.contains(kind, consumer_id) + { Self::send_partition_deny_or_log( &self.consensus, header, @@ -2578,7 +2599,6 @@ where || self .consumer_offset_capacity_for(kind) .is_stranded(consumer_id) - || self.durable_consumer_offsets.contains(kind, consumer_id) { Ok(()) } else { @@ -2593,12 +2613,22 @@ where ReplicaLogContext::from_consensus(self.consensus(), PlaneKind::Partitions) } + fn store_offset_range_error(&self, offset: u64) -> Option<IggyError> { + let current = self.stats.current_offset(); + (offset > current || (current == 0 && self.stats.messages_count_inconsistent() == 0)) + .then_some(IggyError::InvalidOffset(offset)) + } + fn resynchronize_consumer_offset_reservations(&mut self) { let current_view = self.consensus.view(); if current_view == self.observed_view && !self.offset_reservations_need_resync.get() { return; } + if current_view != self.observed_view { + self.queued_auto_commit_reservations.borrow_mut().clear(); + } + let from_op = self .consensus .commit_min() @@ -2665,7 +2695,29 @@ where } self.observed_view = current_view; self.consumer_group_offsets_need_reconcile.set(true); - self.offset_reservations_need_resync.set(decode_failed); + // Repair and truncation rearm this flag when journal contents change. + // A failed decode alone must not cause a full scan on every tick. + self.offset_reservations_need_resync.set(false); + if !decode_failed && self.consensus.is_primary() { + self.reclaim_phantom_offsets(ConsumerKind::Consumer); + self.reclaim_phantom_offsets(ConsumerKind::ConsumerGroup); + } + } + + fn reclaim_phantom_offsets(&self, kind: ConsumerKind) { + let capacity = self.consumer_offset_capacity_for(kind); + let keep = |id| capacity.protects(id, &self.durable_consumer_offsets); + match kind { + ConsumerKind::Consumer => self + .consumer_offsets + .pin() + .retain(|id, _| u32::try_from(*id).ok().is_none_or(keep)), + ConsumerKind::ConsumerGroup => self + .consumer_group_offsets + .pin() + .retain(|id, _| u32::try_from(id.0).ok().is_none_or(keep)), + } + capacity.rearm_map_if_below_limit(self.consumer_offset_map_count(kind)); } /// Build an owned [`PollPlan`] synchronously (no `.await`), so the caller @@ -2724,6 +2776,30 @@ where } }; + if args.auto_commit + && self.consensus.is_primary() + && let Ok(pending) = PendingConsumerOffsetCommit::try_from_polling_consumer(consumer, 0) + { + let capacity = self.consumer_offset_capacity_for(pending.kind); + if !capacity.is_uncertain() + && self.consumer_offset_map_count(pending.kind) >= capacity.limit() + { + let exists = match pending.kind { + ConsumerKind::Consumer => self + .consumer_offsets + .pin() + .contains_key(&(pending.consumer_id as usize)), + ConsumerKind::ConsumerGroup => self + .consumer_group_offsets + .pin() + .contains_key(&ConsumerGroupId(pending.consumer_id as usize)), + }; + if !exists { + self.reclaim_phantom_offsets(pending.kind); + } + } + } + // Past the empty-return guards: only now build the auto-commit context, // whose offset-path `format!()` is wasted on the early returns above. let auto_commit = self.auto_commit_ctx(consumer, args.auto_commit); @@ -3109,6 +3185,23 @@ where message: Message<RoutedRequestHeader>, reply: Option<consensus::Sender<Message<ReplyHeader>>>, ) { + self.on_request_with_reservation(message, reply, None).await; + } + + #[allow(clippy::too_many_lines)] + pub(crate) async fn on_request_with_reservation( + &mut self, + message: Message<RoutedRequestHeader>, + reply: Option<consensus::Sender<Message<ReplyHeader>>>, + mut reservation: Option<crate::AutoCommitReservation>, + ) { + if reservation.as_ref().is_some_and(|reservation| { + !self + .consumer_offset_capacity_for(reservation.kind) + .owns(reservation) + }) { + return; + } // Taken by whichever arm answers: the deny paths, the NoAck fast path, // or the pipeline entry that fires it at commit. Exactly one runs. let mut reply = reply; @@ -3303,30 +3396,26 @@ where // code on this committed-shaped frame (op=commit_max) as success. if matches!(message.header().operation, Operation::StoreConsumerOffset) && let Some((_, _, Some(requested_offset), _)) = consumer_offset + && let Some(error) = self.store_offset_range_error(requested_offset) { - let current_offset = self.stats.current_offset(); - let partition_empty = - self.stats.messages_count_inconsistent() == 0 && current_offset == 0; - if partition_empty || requested_offset > current_offset { - emit_partition_diag( - tracing::Level::WARN, - &PartitionDiagEvent::new( - ReplicaLogContext::from_consensus(consensus, PlaneKind::Partitions), - "rejecting store_consumer_offset for out-of-range offset", - ) - .with_operation(message.header().operation) - .with_error(IggyError::InvalidOffset(requested_offset).to_string()), - ); - Self::send_partition_deny_or_log( - consensus, - message.header(), - IggyError::InvalidOffset(requested_offset).as_code(), - "store_consumer_offset deny reply send failed", - reply.take(), + emit_partition_diag( + tracing::Level::WARN, + &PartitionDiagEvent::new( + ReplicaLogContext::from_consensus(consensus, PlaneKind::Partitions), + "rejecting store_consumer_offset for out-of-range offset", ) - .await; - return; - } + .with_operation(message.header().operation) + .with_error(error.to_string()), + ); + Self::send_partition_deny_or_log( + consensus, + message.header(), + error.as_code(), + "store_consumer_offset deny reply send failed", + reply.take(), + ) + .await; + return; } // The node-local fast path is safe only for a single-replica @@ -3392,6 +3481,12 @@ where waiter, ) .await; + } else if let Some(reservation) = reservation.take() { + self.queued_auto_commit_reservations + .borrow_mut() + .entry((reservation.kind, reservation.consumer_id)) + .or_default() + .push(reservation); } return; } @@ -3471,6 +3566,23 @@ where while promoted < slots_freed { let req = self.consensus().pop_queued_request(); let Some(mut req) = req else { break }; + let _reservation = + Self::parse_consumer_offset_request(req.message.header().operation, &req.message) + .ok() + .and_then(|(kind, id, _, _)| { + if req.message.header().operation != Operation::StoreConsumerOffset + || !is_auto_commit_client(req.message.header().client) + { + return None; + } + let mut queued = self.queued_auto_commit_reservations.borrow_mut(); + let reservations = queued.get_mut(&(kind, id))?; + let reservation = reservations.pop(); + if reservations.is_empty() { + queued.remove(&(kind, id)); + } + reservation + }); // Taken before the preflight so a refusal answers the parked waiter // instead of waking it with `Canceled`. @@ -3498,14 +3610,11 @@ where .await; continue; }; - let current_offset = self.stats.current_offset(); - if offset > current_offset - || (current_offset == 0 && self.stats.messages_count_inconsistent() == 0) - { + if let Some(error) = self.store_offset_range_error(offset) { Self::send_partition_deny_or_log( self.consensus(), req.message.header(), - IggyError::InvalidOffset(offset).as_code(), + error.as_code(), "queued offset range deny reply send failed", reply_sender.take(), ) @@ -5056,6 +5165,9 @@ where send_fail_label: &'static str, waiter: Option<consensus::Sender<Message<ReplyHeader>>>, ) { + if waiter.is_none() && is_auto_commit_client(header.client) { + return; + } let reply = build_deny_reply_from_request(consensus, header, status); Self::deliver_reply_or_log(consensus, header, reply, waiter, send_fail_label).await; } @@ -6105,10 +6217,7 @@ where crate::state_transfer::strayed_offset_files(self.consumer_offsets_path.as_deref(), &[]) .into_iter() .filter_map(|path| { - std::path::Path::new(&path) - .file_name() - .and_then(|name| name.to_str()) - .and_then(|name| name.parse().ok()) + crate::state_transfer::numeric_offset_id(&path) .map(|id| (ConsumerKind::Consumer, id, path)) }); let strayed_groups = crate::state_transfer::strayed_offset_files( @@ -6117,10 +6226,7 @@ where ) .into_iter() .filter_map(|path| { - std::path::Path::new(&path) - .file_name() - .and_then(|name| name.to_str()) - .and_then(|name| name.parse().ok()) + crate::state_transfer::numeric_offset_id(&path) .map(|id| (ConsumerKind::ConsumerGroup, id, path)) }); for (kind, consumer_id, path) in consumer_paths @@ -6175,6 +6281,7 @@ where } self.durable_consumer_offsets.clear(); self.pending_consumer_offset_commits.clear(); + self.queued_auto_commit_reservations.borrow_mut().clear(); self.consumer_offset_capacity .rebuild(&self.durable_consumer_offsets, std::iter::empty()); self.consumer_group_offset_capacity @@ -8396,6 +8503,220 @@ mod tests { ); } + #[compio::test] + async fn given_queued_auto_commit_when_view_changes_should_release_its_provisional_slot() { + let namespace = IggyNamespace::new(1, 1, 0); + let consensus = VsrConsensus::new( + TEST_CLUSTER, + 0, + 3, + namespace.inner(), + RecordingBus::default(), + LocalPipeline::with_capacities(1, 2), + ); + consensus.init(); + let mut partition: IggyPartition<RecordingBus> = IggyPartition::with_in_memory_storage( + Arc::new(PartitionStats::default()), + consensus, + IggyByteSize::from(1024 * 1024), + false, + ); + partition.stats.increment_messages_count(1); + partition.set_consumer_offsets_max(2); + partition + .on_request( + store_offset_request(42, 1, ConsumerKind::Consumer, 7, 0, AckLevel::Quorum), + None, + ) + .await; + let reservation = partition + .consumer_offset_capacity + .reserve_provisional(8, &partition.durable_consumer_offsets) + .unwrap(); + partition + .on_request_with_reservation( + store_offset_request( + message_bus::AUTO_COMMIT_CLIENT_ID, + 1, + ConsumerKind::Consumer, + 8, + 0, + AckLevel::Quorum, + ), + None, + Some(reservation), + ) + .await; + assert_eq!( + partition.occupied_consumer_offset_count(ConsumerKind::Consumer), + 2 + ); + assert_eq!(partition.queued_auto_commit_reservations.borrow().len(), 1); + partition.consensus.set_view(3); + partition.resynchronize_consumer_offset_reservations(); + assert!( + partition + .queued_auto_commit_reservations + .borrow() + .is_empty() + ); + assert_eq!( + partition.occupied_consumer_offset_count(ConsumerKind::Consumer), + 1 + ); + } + + #[compio::test] + async fn given_auto_commit_guard_when_pump_admits_should_transfer_to_journal_without_reply() { + let (mut partition, sent) = recording_partition_at(0, 3); + partition.stats.increment_messages_count(1); + partition.set_consumer_offsets_max(1); + let reservation = partition + .consumer_offset_capacity + .reserve_provisional(7, &partition.durable_consumer_offsets) + .unwrap(); + partition + .on_request_with_reservation( + store_offset_request( + message_bus::AUTO_COMMIT_CLIENT_ID, + 1, + ConsumerKind::Consumer, + 7, + 0, + AckLevel::Quorum, + ), + None, + Some(reservation), + ) + .await; + assert_eq!(partition.pending_consumer_offset_commits.len(), 1); + assert_eq!( + partition.occupied_consumer_offset_count(ConsumerKind::Consumer), + 1 + ); + assert!(sent.borrow().is_empty()); + partition + .apply_staged_consumer_offset_commit(1) + .await + .unwrap(); + assert_eq!( + partition.durable_consumer_offset_count(ConsumerKind::Consumer), + 1 + ); + assert!(sent.borrow().is_empty()); + } + + #[compio::test] + async fn given_group_offset_updates_when_key_already_exists_should_keep_reconciliation_idle() { + let (mut partition, _) = recording_partition(); + partition.stage_consumer_offset_upsert(1, ConsumerKind::ConsumerGroup, 7, 1, true); + partition + .apply_staged_consumer_offset_commit(1) + .await + .unwrap(); + assert!(partition.consumer_group_offsets_reconcile_needed()); + assert!( + partition + .dead_consumer_group_offset_ids(|_| true) + .is_empty() + ); + assert!(!partition.consumer_group_offsets_reconcile_needed()); + partition.stage_consumer_offset_upsert(2, ConsumerKind::ConsumerGroup, 7, 2, true); + partition + .apply_staged_consumer_offset_commit(2) + .await + .unwrap(); + assert!(!partition.consumer_group_offsets_reconcile_needed()); + partition.consumer_group_offsets.pin().insert( + ConsumerGroupId(8), + ConsumerOffset::new(ConsumerKind::ConsumerGroup, 8, 3, String::new()), + ); + partition.stage_consumer_offset_upsert(3, ConsumerKind::ConsumerGroup, 8, 3, true); + partition + .apply_staged_consumer_offset_commit(3) + .await + .unwrap(); + assert!( + partition.consumer_group_offsets_reconcile_needed(), + "first durable commit must discover an eager map key" + ); + } + + #[compio::test] + async fn given_primary_view_change_when_phantoms_exist_should_keep_only_protected_keys() { + let (mut partition, _) = recording_partition_at(0, 3); + partition.set_consumer_offsets_max(4); + partition.stats.increment_messages_count(1); + partition.seed_recovered_consumer_offset(ConsumerKind::Consumer, 7, 0, 0); + partition + .on_request( + store_offset_request(42, 1, ConsumerKind::Consumer, 8, 0, AckLevel::Quorum), + None, + ) + .await; + let held = partition + .consumer_offset_capacity + .reserve_provisional(9, &partition.durable_consumer_offsets) + .unwrap(); + for id in 7..=10 { + partition.consumer_offsets.pin().insert( + id as usize, + ConsumerOffset::new(ConsumerKind::Consumer, id, 0, String::new()), + ); + } + partition.consensus.set_view(3); + partition.resynchronize_consumer_offset_reservations(); + assert_eq!(partition.consumer_offsets.len(), 3); + assert!(!partition.consumer_offsets.pin().contains_key(&10)); + for id in 7..=9 { + assert!(partition.consumer_offsets.pin().contains_key(&id)); + } + drop(held); + } + + #[test] + fn given_full_live_map_when_polling_existing_key_should_reclaim_only_for_new_keys() { + let (mut partition, _) = recording_partition(); + partition.set_consumer_offsets_max(2); + partition.offset_space.committed_seeded = true; + partition.seed_recovered_consumer_offset(ConsumerKind::Consumer, 7, 0, 0); + for id in 7..=8 { + partition.consumer_offsets.pin().insert( + id as usize, + ConsumerOffset::new(ConsumerKind::Consumer, id, 0, String::new()), + ); + } + let args = PollingArgs::new(iggy_common::PollingStrategy::first(), 1, true); + let _ = partition.build_poll_plan(PollingConsumer::Consumer(7, 0), &args, false); + assert_eq!(partition.consumer_offsets.len(), 2); + let _ = partition.build_poll_plan(PollingConsumer::Consumer(9, 0), &args, false); + assert_eq!(partition.consumer_offsets.len(), 1); + assert!(partition.consumer_offsets.pin().contains_key(&7)); + } + + #[test] + fn given_missing_retained_header_when_resync_fails_should_wait_for_new_journal_state() { + let (mut partition, _) = recording_partition(); + partition.consensus.sequencer().set_sequence(1); + partition.consensus.set_view(1); + partition.resynchronize_consumer_offset_reservations(); + assert!(partition.consumer_offset_capacity.is_uncertain()); + assert!(!partition.offset_reservations_need_resync.get()); + partition.seed_recovered_consumer_offset(ConsumerKind::Consumer, 7, 0, 0); + let existing = partition + .auto_commit_ctx(PollingConsumer::Consumer(7, 0), true) + .unwrap() + .apply(0) + .unwrap(); + assert!(partition.auto_commit_admission_ready(&existing)); + let new = partition + .auto_commit_ctx(PollingConsumer::Consumer(8, 0), true) + .unwrap() + .apply(0) + .unwrap(); + assert!(!partition.auto_commit_admission_ready(&new)); + } + #[compio::test] async fn given_full_offset_table_when_storing_new_id_should_reject_before_replication() { let (mut partition, sent_to_clients) = recording_partition(); diff --git a/core/partitions/src/iggy_partitions.rs b/core/partitions/src/iggy_partitions.rs index 2a2ee6e1e..d66b96dc7 100644 --- a/core/partitions/src/iggy_partitions.rs +++ b/core/partitions/src/iggy_partitions.rs @@ -628,6 +628,22 @@ where let _ = waiter.send(build_deny_reply_from_request_header(header, status)); } } + + pub async fn on_auto_commit_request( + &self, + request: Message<RoutedRequestHeader>, + reservation: crate::AutoCommitReservation, + ) { + let namespace = IggyNamespace::from_raw(request.header().group); + if self.is_tombstoned(&namespace) { + return; + } + if let Some(partition) = self.get_mut_by_ns(&namespace) { + partition + .on_request_with_reservation(request, None, Some(reservation)) + .await; + } + } } impl<B, SB> Plane<VsrConsensus<B>> for IggyPartitions<B, SB> diff --git a/core/partitions/src/poll_plan.rs b/core/partitions/src/poll_plan.rs index 176fc8245..51ae28a28 100644 --- a/core/partitions/src/poll_plan.rs +++ b/core/partitions/src/poll_plan.rs @@ -311,6 +311,9 @@ impl PollPlan { /// # Errors /// Returns a capacity error when auto-commit would create a new live-map /// entry after the configured per-kind bound has been reached. + /// The serving shard also rejects completion with `TransientNotAccepted` + /// if the partition disappeared or changed incarnation during disk I/O. + /// No fragments are returned for that rejected completion. pub async fn execute( self, ) -> Result<(PollFragments<4096>, u64, Option<AutoCommitApplied>), ConsumerOffsetCapacityError> @@ -890,64 +893,37 @@ impl AutoCommitCtx { offset: u64, ) -> Result<AutoCommitApplied, ConsumerOffsetCapacityError> { let (kind, consumer_id) = self.kind_and_id(); + let create = |path: Option<&str>| { + ConsumerOffset::new( + kind, + consumer_id, + offset, + path.map_or_else(String::new, |path| format!("{path}/{consumer_id}")), + ) + }; let previous_offset = match &self.target { AutoCommitTarget::Consumer { offsets, consumer_id, create_path, - } => { - let consumer_id = *consumer_id; - let map: &ConsumerOffsets = offsets; - let guard = map.pin(); - if let Some(existing) = guard.get(&(consumer_id as usize)) { - Some(existing.offset.fetch_max(offset, Ordering::Relaxed)) - } else { - self.capacity.admit_local_map_key(guard.len())?; - let created = create_path.as_deref().map_or_else( - || { - ConsumerOffset::new( - ConsumerKind::Consumer, - consumer_id, - offset, - String::new(), - ) - }, - |path| ConsumerOffset::default_for_consumer(consumer_id, path), - ); - created.offset.store(offset, Ordering::Relaxed); - guard.insert(consumer_id as usize, created); - None - } - } + } => apply_local_offset( + offsets, + *consumer_id as usize, + offset, + &self.capacity, + || create(create_path.as_deref()), + )?, AutoCommitTarget::ConsumerGroup { offsets, group_id, create_path, - } => { - let group_id = *group_id; - let key = ConsumerGroupId(group_id as usize); - let map: &ConsumerGroupOffsets = offsets; - let guard = map.pin(); - if let Some(existing) = guard.get(&key) { - Some(existing.offset.fetch_max(offset, Ordering::Relaxed)) - } else { - self.capacity.admit_local_map_key(guard.len())?; - let created = create_path.as_deref().map_or_else( - || { - ConsumerOffset::new( - ConsumerKind::ConsumerGroup, - group_id, - offset, - String::new(), - ) - }, - |path| ConsumerOffset::default_for_consumer_group(key, path), - ); - created.offset.store(offset, Ordering::Relaxed); - guard.insert(key, created); - None - } - } + } => apply_local_offset( + offsets, + ConsumerGroupId(*group_id as usize), + offset, + &self.capacity, + || create(create_path.as_deref()), + )?, }; Ok(AutoCommitApplied { kind, @@ -980,13 +956,7 @@ impl AutoCommitApplied { ) -> Result<Option<AutoCommitReservation>, ConsumerOffsetCapacityError> { if self .durable - .get(self.kind, self.consumer_id) - .is_some_and(|state| { - state.committed_offset >= self.offset - && state - .persisted_high_water - .is_some_and(|high_water| self.offset <= high_water) - }) + .covers(self.kind, self.consumer_id, self.offset) { return Ok(None); } @@ -1007,35 +977,51 @@ impl AutoCommitApplied { offsets, consumer_id, .. - } => { - let guard = offsets.pin(); - if let Some(previous) = self.previous_offset { - if let Some(entry) = guard.get(&(*consumer_id as usize)) { - entry.offset.store(previous, Ordering::Relaxed); - } - } else { - guard.remove(&(*consumer_id as usize)); - } - guard.len() - } + } => rollback_local_offset(offsets, *consumer_id as usize, self.previous_offset), AutoCommitTarget::ConsumerGroup { offsets, group_id, .. - } => { - let guard = offsets.pin(); - if let Some(previous) = self.previous_offset { - if let Some(entry) = guard.get(&ConsumerGroupId(*group_id as usize)) { - entry.offset.store(previous, Ordering::Relaxed); - } - } else { - guard.remove(&ConsumerGroupId(*group_id as usize)); - } - guard.len() - } + } => rollback_local_offset( + offsets, + ConsumerGroupId(*group_id as usize), + self.previous_offset, + ), }; self.capacity.rearm_map_if_below_limit(map_len); } } +fn rollback_local_offset<K: Hash + Eq + Send + Sync + Copy>( + map: &papaya::HashMap<K, ConsumerOffset>, + key: K, + previous: Option<u64>, +) -> usize { + let guard = map.pin(); + if let Some(previous) = previous { + if let Some(entry) = guard.get(&key) { + entry.offset.store(previous, Ordering::Relaxed); + } + } else { + guard.remove(&key); + } + guard.len() +} + +fn apply_local_offset<K: Hash + Eq + Clone + Send + Sync>( + map: &papaya::HashMap<K, ConsumerOffset>, + key: K, + offset: u64, + capacity: &ConsumerOffsetCapacity, + create: impl FnOnce() -> ConsumerOffset, +) -> Result<Option<u64>, ConsumerOffsetCapacityError> { + let guard = map.pin(); + if let Some(existing) = guard.get(&key) { + return Ok(Some(existing.offset.fetch_max(offset, Ordering::Relaxed))); + } + capacity.admit_local_map_key(guard.len())?; + guard.insert(key, create()); + Ok(None) +} + /// Upsert a committed offset into a lock-free `papaya` offset map: bump an /// existing entry in place, or build one via `create_on_miss` on first commit /// for a consumer/group that has none yet. Shared by the pump's diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs index ff5efe345..41d7f6f09 100644 --- a/core/partitions/src/state_transfer.rs +++ b/core/partitions/src/state_transfer.rs @@ -2000,12 +2000,9 @@ where let mut consumers = self .durable_consumer_offsets .committed_entries(ConsumerKind::Consumer); + let consumer_map = self.consumer_offsets.pin(); for (consumer_id, _) in &consumers { - if !self - .consumer_offsets - .pin() - .contains_key(&(*consumer_id as usize)) - { + if !consumer_map.contains_key(&(*consumer_id as usize)) { return Err( PartitionTransferUnavailable::ConsumerOffsetStateInconsistent { kind: ConsumerKind::Consumer, @@ -2018,12 +2015,9 @@ where let mut groups = self .durable_consumer_offsets .committed_entries(ConsumerKind::ConsumerGroup); + let group_map = self.consumer_group_offsets.pin(); for (consumer_id, _) in &groups { - if !self - .consumer_group_offsets - .pin() - .contains_key(&ConsumerGroupId(*consumer_id as usize)) - { + if !group_map.contains_key(&ConsumerGroupId(*consumer_id as usize)) { return Err( PartitionTransferUnavailable::ConsumerOffsetStateInconsistent { kind: ConsumerKind::ConsumerGroup, @@ -2825,6 +2819,7 @@ where } self.durable_consumer_offsets.clear(); self.pending_consumer_offset_commits.clear(); + self.queued_auto_commit_reservations.borrow_mut().clear(); self.consumer_offset_capacity .rebuild(&self.durable_consumer_offsets, std::iter::empty()); self.consumer_group_offset_capacity @@ -3450,7 +3445,7 @@ pub(crate) fn strayed_offset_files(dir: Option<&str>, incoming: &[(u32, u64)]) - .collect() } -fn numeric_offset_id(path: &str) -> Option<u32> { +pub(crate) fn numeric_offset_id(path: &str) -> Option<u32> { Path::new(path).file_name()?.to_str()?.parse().ok() } diff --git a/core/sdk/Cargo.toml b/core/sdk/Cargo.toml index 09ef84d03..93993589c 100644 --- a/core/sdk/Cargo.toml +++ b/core/sdk/Cargo.toml @@ -17,7 +17,7 @@ [package] name = "iggy" -version = "0.11.0-edge.6" +version = "0.11.0-edge.7" 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/clients/consumer.rs b/core/sdk/src/clients/consumer.rs index 1f38f2c76..e957cce7a 100644 --- a/core/sdk/src/clients/consumer.rs +++ b/core/sdk/src/clients/consumer.rs @@ -618,6 +618,11 @@ unsafe impl Sync for IggyConsumer {} /// [`topic()`]: crate::prelude::IggyConsumerBuilder::topic /// [`without_encryptor()`]: crate::prelude::IggyConsumerBuilder::without_encryptor /// [`without_poll_interval()`]: crate::prelude::IggyConsumerBuilder::without_poll_interval +/// +/// A server-side auto-commit poll can fail with `TooManyConsumerOffsets` when +/// its consumer needs a new offset key at the partition's configured limit. +/// The rejected poll returns no messages. Existing offset keys remain usable, +/// and polling with server-side auto-commit disabled allocates no offset key. pub struct IggyConsumer { initialized: bool, shutdown: Arc<AtomicBool>, diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml index 1d5911488..553d8d9ff 100644 --- a/core/server/Cargo.toml +++ b/core/server/Cargo.toml @@ -17,7 +17,7 @@ [package] name = "server" -version = "0.9.0-edge.6" +version = "0.9.0-edge.7" edition = "2024" license = "Apache-2.0" publish = false diff --git a/core/server/config.toml b/core/server/config.toml index 285e9c807..909b791c5 100644 --- a/core/server/config.toml +++ b/core/server/config.toml @@ -1027,6 +1027,9 @@ dedup_clients_max = 4096 # Distinct durable consumer-offset keys a partition primary admits per kind. # Standalone consumers and consumer groups are counted separately. Existing # keys remain writable at the limit. A new key is rejected before consensus. +# A non-empty auto_commit poll that needs a new offset key is also rejected +# with TooManyConsumerOffsets and returns no messages. Polls with auto_commit +# disabled do not allocate offset keys and remain available. # Must be > 0 and <= 262144. consumer_offsets_max = 4096 diff --git a/core/server/src/boot/recovery.rs b/core/server/src/boot/recovery.rs index 0610e826a..917306bdc 100644 --- a/core/server/src/boot/recovery.rs +++ b/core/server/src/boot/recovery.rs @@ -336,7 +336,7 @@ const _: () = assert!( == partitions::DEFAULT_CONSUMER_OFFSETS_MAX ); const _: () = assert!( - 2 * configs::partition::PARTITION_CONSUMER_OFFSETS_CEILING + 4 * configs::partition::PARTITION_CONSUMER_OFFSETS_CEILING <= partitions::CONSUMER_OFFSETS_ENTRIES_MAX as usize ); const _: () = diff --git a/core/server/src/consumer_group.rs b/core/server/src/consumer_group.rs index 2f347782f..1a6beed9b 100644 --- a/core/server/src/consumer_group.rs +++ b/core/server/src/consumer_group.rs @@ -25,7 +25,7 @@ //! primary enriches the op here before replication, mirroring the PAT mint //! in [`crate::pat`] and the password hash in [`crate::users`]. -use crate::responses::resolve_partition_namespace; +use crate::responses::{missing_consumer_group_error, resolve_partition_namespace}; use crate::shell::{ShellBus, ShellShard}; use crate::wire::{request_body, rewrite_request_body}; use consensus::MetadataHandle; @@ -235,9 +235,12 @@ where if wire.consumer.kind != KIND_CONSUMER_GROUP { return Ok(request); } - let group_id = - resolve_group_offset_id(shard, &wire.consumer, (&wire.stream_id, &wire.topic_id)) - .ok_or(IggyError::InvalidIdentifier)?; + let group_id = resolve_group_offset_id( + shard, + &wire.consumer, + (&wire.stream_id, &wire.topic_id), + ) + .ok_or_else(|| missing_consumer_group_error(&wire.consumer.id, &wire.topic_id))?; // The partition-plane group-offset key is u32 (see the documented // ceiling on `Topic::next_consumer_group_id`). Clamp on the // ~4-billion-creates overflow rather than panic this live @@ -270,9 +273,6 @@ where S: 'static, SB: SuperblockStore + 'static, { - if consumer.kind != KIND_CONSUMER_GROUP { - return None; - } shard .plane .metadata() diff --git a/core/server/src/dispatch/partition.rs b/core/server/src/dispatch/partition.rs index 1c8e21711..48bbe0b5b 100644 --- a/core/server/src/dispatch/partition.rs +++ b/core/server/src/dispatch/partition.rs @@ -51,7 +51,7 @@ use iggy_binary_protocol::PrepareHeader; use iggy_binary_protocol::primitives::consumer::WireConsumer; use iggy_binary_protocol::primitives::polling_strategy::WirePollingStrategy; use iggy_binary_protocol::requests::consumer_offsets::{ - DeleteConsumerOffsetRequest, GetConsumerOffsetRequest, StoreConsumerOffsetRequest, + GetConsumerOffsetRequest, StoreConsumerOffsetRequest, }; use iggy_binary_protocol::requests::messages::PollMessagesRequest; use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest; @@ -128,7 +128,7 @@ where Err(error) => { shard.metrics().record_consumer_offset_denied(error.kind); warn_auto_commit_capacity(namespace, error); - PartitionReadReply::Rejected(IggyError::TooManyConsumerOffsets) + PartitionReadReply::Rejected(error.into()) } }; let _ = reply.try_send(result); @@ -228,7 +228,7 @@ fn spawn_poll_io<B, MJ, S, SB>( Err(error) => { shard.metrics().record_consumer_offset_denied(error.kind); warn_auto_commit_capacity(namespace, error); - PartitionReadReply::Rejected(IggyError::TooManyConsumerOffsets) + PartitionReadReply::Rejected(error.into()) } }; let _ = reply.try_send(result); @@ -296,7 +296,7 @@ where shard.metrics().record_consumer_offset_denied(applied.kind); warn_auto_commit_capacity(namespace, error); applied.rollback_created(); - return Err(IggyError::TooManyConsumerOffsets); + return Err(error.into()); } }; let message = match build_auto_commit_request(namespace, applied) { @@ -313,15 +313,13 @@ where }; // Routes by namespace to this same owning primary shard's inbox. The pump // admits it next turn exactly like a client store. `dispatch` never blocks. - let Ok(ticket) = shard.partition_submit(namespace, message) else { + if shard + .submit_auto_commit_offset(message, reservation) + .is_err() + { applied.rollback_created(); return Err(IggyError::TransientNotAccepted); - }; - let shard = Rc::clone(shard); - shard.bus.clone().spawn(async move { - let _ = shard.await_partition_submit(ticket).await; - drop(reservation); - }); + } applied.mark_served(); Ok(()) } @@ -449,7 +447,10 @@ pub async fn dispatch_partition_request<B, MJ, S, SB>( operation = ?header.operation, "partition request with unresolved namespace; replying denied" ); - let status = if error == IggyError::InvalidIdentifier { + let status = if matches!( + error, + IggyError::ConsumerGroupIdNotFound(..) | IggyError::ConsumerGroupNameNotFound(..) + ) { error.as_code() } else { IggyError::ResourceNotFound(String::new()).as_code() @@ -650,22 +651,12 @@ async fn relay_partition_reply<B, MJ, S, SB>( } fn consumer_offset_kind(request: &Message<RoutedRequestHeader>) -> Option<ConsumerKind> { - let kind = match request.header().operation { - Operation::StoreConsumerOffset => { - StoreConsumerOffsetRequest::decode_from(request_body(request)) - .ok()? - .consumer - .kind - } - Operation::DeleteConsumerOffset => { - DeleteConsumerOffsetRequest::decode_from(request_body(request)) - .ok()? - .consumer - .kind - } - _ => return None, - }; - ConsumerKind::from_code(kind).ok() + if request.header().operation != Operation::StoreConsumerOffset { + return None; + } + // WireConsumer starts with its kind byte. The dispatch path has already + // decoded and validated the complete request. + ConsumerKind::from_code(*request_body(request).first()?).ok() } /// Serve `poll_messages`: resolve the partition namespace, run the read on diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs index b8fc36c3d..6c3f888c7 100644 --- a/core/server/src/partition_helpers.rs +++ b/core/server/src/partition_helpers.rs @@ -276,10 +276,16 @@ pub fn configure_consumer_offsets( enforce_fsync, ); for consumer_id in numeric_offset_file_ids(&consumer_offsets_path) { - partition.seed_stranded_consumer_offset(ConsumerKind::Consumer, consumer_id); + if partition.seed_stranded_consumer_offset(ConsumerKind::Consumer, consumer_id) { + warn!(stream_id, topic_id, partition_id, consumer_id, path = %consumer_offsets_path, + "unloaded consumer offset file retains its capacity slot until updated or deleted"); + } } for group_id in numeric_offset_file_ids(&consumer_group_offsets_path) { - partition.seed_stranded_consumer_offset(ConsumerKind::ConsumerGroup, group_id); + if partition.seed_stranded_consumer_offset(ConsumerKind::ConsumerGroup, group_id) { + warn!(stream_id, topic_id, partition_id, group_id, path = %consumer_group_offsets_path, + "unloaded group offset file retains its capacity slot until repaired or reclaimed"); + } } for kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] { let count = partition.occupied_consumer_offset_count(kind); diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs index 21f018085..8fedc7204 100644 --- a/core/server/src/partition_reconciler.rs +++ b/core/server/src/partition_reconciler.rs @@ -194,6 +194,7 @@ use tracing::{debug, error, trace}; const BACKOFF_BASE: Duration = Duration::from_secs(1); const BACKOFF_MAX: Duration = Duration::from_mins(1); +const GROUP_OFFSET_DELETES_PER_PASS: usize = 32; /// Consecutive same-cause failures before [`ReconcilerCtx::record_failure`] /// escalates to an operator-visible error (the backoff is capped, so @@ -235,6 +236,8 @@ pub struct ReconcilerCtx { /// `true` when the previous pass made no changes. Only then is a /// same-`revision` pass safe to skip. last_pass_noop: Cell<bool>, + group_offset_cleanup_inflight: Rc<RefCell<AHashSet<IggyNamespace>>>, + group_offset_cleanup_completed: Rc<Cell<usize>>, } impl ReconcilerCtx { @@ -257,6 +260,8 @@ impl ReconcilerCtx { failure_state: RefCell::new(AHashMap::new()), last_revision: Cell::new(None), last_pass_noop: Cell::new(false), + group_offset_cleanup_inflight: Rc::new(RefCell::new(AHashSet::new())), + group_offset_cleanup_completed: Rc::new(Cell::new(0)), } } @@ -389,11 +394,11 @@ struct PassCounters { backoff_skipped: usize, /// Stale incarnations (slab-key reuse) torn down for rebuild. stale: usize, - /// Consumer-group offsets reclaimed for groups deleted while their topic - /// survived (a bare `DeleteConsumerGroup`, not a topic/stream delete). - cg_offsets_purged: usize, - /// Consumer-group offset files whose unlink failed and remain queued for - /// the next pass. Counted so the revision fast-skip cannot strand them. + /// Group-offset deletes successfully handed to the pump this pass. + cg_offsets_submitted: usize, + /// Successful replicated deletes reported by detached completion tasks. + cg_offsets_completed: usize, + /// Group-offset deletes refused by the inbox and needing another pass. cg_offsets_pending: usize, /// Committed delete watermarks not yet fully enforced on local segments. /// Counted so the pass does not arm the fast-skip: the pump can be @@ -432,7 +437,8 @@ impl PassCounters { + self.removed_routed + self.backoff_skipped + self.stale - + self.cg_offsets_purged + + self.cg_offsets_submitted + + self.cg_offsets_completed + self.cg_offsets_pending + self.trims_pending + self.purges_staged @@ -474,6 +480,7 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool { if ctx.last_revision.get() == Some(revision) && ctx.last_pass_noop.get() && ctx.failure_state.borrow().is_empty() + && ctx.group_offset_cleanup_completed.get() == 0 && !ctx.shard.has_parked_partition_frames() && !ctx.shard.plane.partitions().namespaces().any(|namespace| { ctx.shard @@ -481,6 +488,10 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool { .partitions() .with_partition(namespace, |partition| { partition.consumer_group_offsets_reconcile_needed() + && !ctx + .group_offset_cleanup_inflight + .borrow() + .contains(namespace) }) .unwrap_or(false) }) @@ -494,12 +505,15 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool { let target = snapshot_target_namespaces(ctx); let target_set: AHashSet<IggyNamespace> = target.iter().map(|partition| partition.ns).collect(); - let mut counters = PassCounters::default(); + let mut counters = PassCounters { + cg_offsets_completed: ctx.group_offset_cleanup_completed.replace(0), + ..PassCounters::default() + }; reconcile_additions(ctx, target, &mut counters).await; reconcile_removals(ctx, &target_set, &mut counters).await; reconcile_parked_frames(ctx, &mut counters); - reconcile_consumer_group_offsets(ctx, &mut counters).await; + reconcile_consumer_group_offsets(ctx, &mut counters); reconcile_segment_truncations(ctx, &mut counters); reconcile_partition_purges(ctx, &mut counters); @@ -982,11 +996,18 @@ async fn tear_down_owned_partition( /// Reclaim deleted groups through ordered offset deletes. Replicas must see /// each delete before a replacement store can reuse its durable slot. -async fn reconcile_consumer_group_offsets(ctx: &ReconcilerCtx, counters: &mut PassCounters) { +fn reconcile_consumer_group_offsets(ctx: &ReconcilerCtx, counters: &mut PassCounters) { let live_groups = snapshot_topic_live_groups(ctx); let partitions = ctx.shard.plane.partitions(); let owned: Vec<IggyNamespace> = partitions.namespaces().copied().collect(); for namespace in owned { + if ctx + .group_offset_cleanup_inflight + .borrow() + .contains(&namespace) + { + continue; + } let live = live_groups.get(&(namespace.stream_id(), namespace.topic_id())); let dead = partitions .with_partition(&namespace, |partition| { @@ -997,32 +1018,48 @@ async fn reconcile_consumer_group_offsets(ctx: &ReconcilerCtx, counters: &mut Pa .unwrap_or_default(); // Bound work per pass so a historical directory cannot monopolize the // reconciler. Unprocessed keys keep the partition's dirty flag armed. - let mut tickets = Vec::with_capacity(dead.len().min(32)); - for consumer_id in dead.into_iter().take(32) { - counters.cg_offsets_pending += 1; + let mut tickets = Vec::with_capacity(dead.len().min(GROUP_OFFSET_DELETES_PER_PASS)); + for consumer_id in dead.into_iter().take(GROUP_OFFSET_DELETES_PER_PASS) { let request = group_offset_delete_request(namespace, consumer_id); if let Ok(ticket) = ctx.shard.partition_submit(namespace, request) { + counters.cg_offsets_submitted += 1; tickets.push(ticket); + } else { + counters.cg_offsets_pending += 1; } } - let replies = futures::future::join_all( - tickets - .into_iter() - .map(|ticket| ctx.shard.await_partition_submit(ticket)), - ) - .await; - for reply in replies { - let Some(reply) = reply else { - continue; - }; - let header = reply - .as_slice() - .get(..size_of::<ReplyHeader>()) - .and_then(|bytes| bytemuck::checked::try_from_bytes::<ReplyHeader>(bytes).ok()); - if header.is_some_and(|header| header.status == 0) { - counters.cg_offsets_purged += 1; - } + if tickets.is_empty() { + continue; } + ctx.group_offset_cleanup_inflight + .borrow_mut() + .insert(namespace); + let inflight = Rc::clone(&ctx.group_offset_cleanup_inflight); + let completed = Rc::clone(&ctx.group_offset_cleanup_completed); + let shard = Rc::clone(&ctx.shard); + shard.bus.clone().spawn(async move { + let replies = futures::future::join_all( + tickets + .into_iter() + .map(|ticket| shard.await_partition_submit(ticket)), + ) + .await; + let count = replies + .into_iter() + .flatten() + .filter(|reply| { + reply + .as_slice() + .get(..size_of::<ReplyHeader>()) + .and_then(|bytes| { + bytemuck::checked::try_from_bytes::<ReplyHeader>(bytes).ok() + }) + .is_some_and(|header| header.status == 0) + }) + .count(); + completed.set(completed.get() + count); + inflight.borrow_mut().remove(&namespace); + }); } } @@ -1331,8 +1368,9 @@ pub fn install_tick_handler(shard: &Rc<ServerShard>, wake_tx: WakeTx) { #[cfg(test)] mod tests { use super::{ - FailureCause, FailureRecord, ReconcilerCtx, build_partition_fresh, - delete_partitions_from_disk, fetch_partition_stats, reconcile_once, + FailureCause, FailureRecord, PassCounters, ReconcilerCtx, build_partition_fresh, + delete_partitions_from_disk, fetch_partition_stats, reconcile_consumer_group_offsets, + reconcile_once, }; use configs::server::{ServerConfig, ServerSystemConfig}; use consensus::{MetadataHandle, PartitionsHandle}; @@ -3241,6 +3279,60 @@ mod tests { /// Failed delivery must leave the offset and its quota slot intact for a /// later replicated delete. This fixture deliberately has no running pump. + #[compio::test] + async fn given_pending_group_cleanup_when_another_topic_arrives_should_reconcile_without_waiting() + { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "cleanup-stream"); + seed_topic(&mux, 2, 0, "cleanup-topic", vec![assignment(0, 1)]); + let (shard, _inbox) = build_test_shard_with_inbox(0, &config, mux, 32); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + reconcile_pass(&ctx).await; + let namespace = IggyNamespace::new(0, 0, 0); + shard + .plane + .partitions() + .with_partition(&namespace, |partition| { + partition.consumer_group_offsets.pin().insert( + iggy_common::ConsumerGroupId(7), + iggy_common::ConsumerOffset::new( + iggy_common::ConsumerKind::ConsumerGroup, + 7, + 0, + String::new(), + ), + ); + }); + reconcile_consumer_group_offsets(&ctx, &mut PassCounters::default()); + assert!( + ctx.group_offset_cleanup_inflight + .borrow() + .contains(&namespace) + ); + seed_topic( + &shard.plane.metadata().mux_stm, + 3, + 0, + "unrelated-topic", + vec![assignment(0, 1)], + ); + reconcile_pass(&ctx).await; + assert!( + shard + .plane + .partitions() + .contains(&IggyNamespace::new(0, 1, 0)), + "an unanswered cleanup must not block partition creation" + ); + assert!( + ctx.group_offset_cleanup_inflight + .borrow() + .contains(&namespace) + ); + } + #[compio::test] async fn given_deleted_group_when_cleanup_submit_fails_should_preserve_state_for_retry() { use iggy_common::{ConsumerGroupId, ConsumerKind, ConsumerOffset}; diff --git a/core/server/src/responses.rs b/core/server/src/responses.rs index cc0654af1..2b9b524a5 100644 --- a/core/server/src/responses.rs +++ b/core/server/src/responses.rs @@ -279,12 +279,6 @@ where } let partition_id = partition_id.ok_or(IggyError::InvalidIdentifier)?; let streams = shard.plane.metadata().mux_stm.streams(); - if streams - .resolve_consumer_group_id(stream_id, topic_id, &consumer.id) - .is_none() - { - return Err(IggyError::InvalidIdentifier); - } #[allow(clippy::cast_possible_truncation)] streams // Commit fence: allow a pending-revoked partition (the source commits it @@ -298,10 +292,28 @@ where false, ) .map(|_| ()) - .ok_or(IggyError::ConsumerGroupPartitionNotOwned( - client_id as u32, - partition_id, - )) + .ok_or_else(|| { + if streams + .resolve_consumer_group_id(stream_id, topic_id, &consumer.id) + .is_none() + { + missing_consumer_group_error(&consumer.id, topic_id) + } else { + IggyError::ConsumerGroupPartitionNotOwned(client_id as u32, partition_id) + } + }) +} + +pub fn missing_consumer_group_error(group: &WireIdentifier, topic: &WireIdentifier) -> IggyError { + let topic = wire_identifier_for_display(topic); + match group { + WireIdentifier::Numeric(_) => { + IggyError::ConsumerGroupIdNotFound(wire_identifier_for_display(group), topic) + } + WireIdentifier::String(name) => { + IggyError::ConsumerGroupNameNotFound(name.as_str().to_owned(), topic) + } + } } /// Fence a consumer-group offset op then resolve its target partition diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 4dbcdf428..2922b8150 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -114,6 +114,7 @@ where pub struct PartitionMaterialisation { epoch: u64, created_view: u32, + consumer_offsets_max: usize, } #[cfg(feature = "simulator")] @@ -123,8 +124,15 @@ impl PartitionMaterialisation { Self { epoch, created_view, + consumer_offsets_max: partitions::DEFAULT_CONSUMER_OFFSETS_MAX, } } + + #[must_use] + pub const fn with_consumer_offsets_max(mut self, consumer_offsets_max: usize) -> Self { + self.consumer_offsets_max = consumer_offsets_max; + self + } } /// Replica id + count bundle. @@ -736,6 +744,12 @@ pub enum LifecycleFrame { request: Message<RoutedRequestHeader>, reply: Sender<Option<Message<GenericHeader>>>, }, + /// Local auto-commit submission. The guard travels with the frame so an + /// inbox drop or admission refusal releases its provisional key directly. + AutoCommitSubmit { + request: Message<RoutedRequestHeader>, + reservation: partitions::AutoCommitReservation, + }, /// Shard 0 broadcasts after a partition-shaped metadata commit; wakes /// the per-shard reconciler. No payload: reconciler re-reads target /// state. Drops covered by the periodic safety tick. @@ -2010,6 +2024,32 @@ where }) } + /// Submit an auto-commit back to the partition-owning shard's pump. + /// + /// # Errors + /// Returns a refusal if the local inbox cannot accept the frame. + pub fn submit_auto_commit_offset( + &self, + request: Message<RoutedRequestHeader>, + reservation: partitions::AutoCommitReservation, + ) -> Result<(), PartitionSubmitRefused> { + let frame = ShardFrame::lifecycle(LifecycleFrame::AutoCommitSubmit { + request, + reservation, + }); + let sender = self + .senders + .get(usize::from(self.id)) + .ok_or(PartitionSubmitRefused)?; + sender.try_send(frame).map_err(|error| { + self.metrics.record_frame_drop( + crate::metrics::frame_drop_variant::PARTITION, + crate::coordinator::classify_try_send_err(&error), + ); + PartitionSubmitRefused + }) + } + /// Wait out a submitted write's committed reply. /// /// `None` = reply channel dropped before a reply (view-change reset, park @@ -3966,7 +4006,6 @@ where // and `adopt_retained_log` are configured out and the crate does not compile. // The feature forwards to `partitions/simulator` instead. #[cfg(feature = "simulator")] - #[allow(clippy::too_many_arguments)] pub fn init_partition( &self, namespace: IggyNamespace, @@ -3975,7 +4014,6 @@ where retained: Option<partitions::RetainedPartitionState>, restore_frontier: bool, materialisation: PartitionMaterialisation, - consumer_offsets_max: usize, ) where B: MessageBus + Clone + 'static, T: ShardsTable, @@ -3983,6 +4021,7 @@ where let PartitionMaterialisation { epoch, created_view, + consumer_offsets_max, } = materialisation; let partitions = self.plane.partitions(); if partitions.contains(&namespace) { diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs index d355b0381..91b6a2001 100644 --- a/core/shard/src/metrics.rs +++ b/core/shard/src/metrics.rs @@ -189,6 +189,13 @@ fn reason_index(s: &str) -> Option<usize> { REASONS.iter().position(|r| *r == s) } +const fn consumer_kind_index(kind: ConsumerKind) -> usize { + match kind { + ConsumerKind::Consumer => 0, + ConsumerKind::ConsumerGroup => 1, + } +} + /// Per-shard metric handles. /// /// Cheap to clone (`Arc` of a `Family` under the hood). Each shard owns @@ -271,21 +278,13 @@ impl ShardMetrics { } pub fn record_consumer_offset_denied(&self, kind: ConsumerKind) { - let index = match kind { - ConsumerKind::Consumer => 0, - ConsumerKind::ConsumerGroup => 1, - }; - self.consumer_offset_denied_counters[index].inc(); + self.consumer_offset_denied_counters[consumer_kind_index(kind)].inc(); } - #[cfg(any(test, feature = "simulator"))] + #[cfg(test)] #[must_use] pub fn consumer_offset_denied_value(&self, kind: ConsumerKind) -> u64 { - let index = match kind { - ConsumerKind::Consumer => 0, - ConsumerKind::ConsumerGroup => 1, - }; - self.consumer_offset_denied_counters[index].get() + self.consumer_offset_denied_counters[consumer_kind_index(kind)].get() } /// Bumped every time a client request is answered with a retryable denial diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs index 4a3a0d214..b7465d2b7 100644 --- a/core/shard/src/router.rs +++ b/core/shard/src/router.rs @@ -728,6 +728,15 @@ where // already made. self.on_partition_submit(request, reply).await; } + LifecycleFrame::AutoCommitSubmit { + request, + reservation, + } => { + self.plane + .partitions() + .on_auto_commit_request(request, reservation) + .await; + } LifecycleFrame::MetadataCommitTick => { // Reconciler may not yet be wired (e.g. mid-bootstrap, or // single-shard tests that never enable the reconciler loop). diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index a4e055c82..242d4004f 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -505,11 +505,6 @@ impl Simulator { self.consumer_offsets_max = consumer_offsets_max; } - #[must_use] - pub const fn consumer_offsets_max(&self) -> usize { - self.consumer_offsets_max - } - #[must_use] /// # Panics /// Panics when `replica_idx` is outside the simulated roster. @@ -1350,8 +1345,10 @@ impl Simulator { }; // Partitions are driven directly, so a poll's auto-commit is never // replicated (the serving shard's job in the real server). Offset discarded. - let (fragments, _commit_offset, _auto_commit) = futures::executor::block_on(plan.execute()) - .map_err(|_| IggyError::TooManyConsumerOffsets)?; + let (fragments, _commit_offset, auto_commit) = futures::executor::block_on(plan.execute())?; + if let Some(applied) = auto_commit { + applied.mark_served(); + } Ok(fragments) } @@ -1527,8 +1524,8 @@ fn materialise_partition( recovered_state, retained, restore_frontier, - PartitionMaterialisation::new(epoch, created_view), - consumer_offsets_max, + PartitionMaterialisation::new(epoch, created_view) + .with_consumer_offsets_max(consumer_offsets_max), ); for shard in &replica.shards { shard.shards_table().insert( @@ -3059,7 +3056,6 @@ mod tests { None, false, PartitionMaterialisation::new(0, 0), - partitions::DEFAULT_CONSUMER_OFFSETS_MAX, ); }); executor.run_until_stalled(POLL_BUDGET); // grow while the borrow is live @@ -3102,7 +3098,6 @@ mod tests { None, false, PartitionMaterialisation::new(0, 0), - partitions::DEFAULT_CONSUMER_OFFSETS_MAX, ); }); executor.run_until_stalled(POLL_BUDGET); diff --git a/core/simulator/src/workload/effect.rs b/core/simulator/src/workload/effect.rs index a9d8c868c..e8df466e9 100644 --- a/core/simulator/src/workload/effect.rs +++ b/core/simulator/src/workload/effect.rs @@ -17,7 +17,7 @@ //! Predicted server-state mutations emitted by op modules on commit. //! -//! Name-keyed throughout. Server-ng emits empty reply bodies, so the +//! Name-keyed throughout. Server emits empty reply bodies, so the //! workload cannot recover server-assigned numeric ids; shadow lookups //! address entities by name (`WireIdentifier::named`). Id-keyed effects //! return once reply-body parsing lands. diff --git a/core/simulator/src/workload/shadow.rs b/core/simulator/src/workload/shadow.rs index 2692ff168..b0e0ea8cb 100644 --- a/core/simulator/src/workload/shadow.rs +++ b/core/simulator/src/workload/shadow.rs @@ -17,7 +17,7 @@ //! Shadow state: the workload's prediction of server-side entity state. //! -//! Name-keyed throughout. Server-ng does not yet ship reply bodies, so +//! Name-keyed throughout. Server does not yet ship reply bodies, so //! the workload cannot observe server-assigned numeric ids. Lookups are //! by name; requests route via `WireIdentifier::named(...)`. When typed //! response bodies land, id-keyed maps return as a parallel index; diff --git a/examples/node/package-lock.json b/examples/node/package-lock.json index fdf8f7883..2847ff873 100644 --- a/examples/node/package-lock.json +++ b/examples/node/package-lock.json @@ -23,7 +23,7 @@ }, "../../foreign/node": { "name": "apache-iggy", - "version": "0.10.0-edge.5", + "version": "0.10.0-edge.6", "license": "Apache-2.0", "dependencies": { "@node-rs/xxhash": "1.7.7", diff --git a/examples/python/uv.lock b/examples/python/uv.lock index bde592d6b..469565c7e 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.dev6" +version = "0.9.0.dev7" source = { directory = "../../foreign/python" } [package.metadata] diff --git a/foreign/go/contracts/version.go b/foreign/go/contracts/version.go index 32ff7a282..f3799086d 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.5" +const Version = "0.9.0-edge.6" diff --git a/foreign/go/errors/errors.yaml b/foreign/go/errors/errors.yaml index 7219af03e..e9f8b7c96 100644 --- a/foreign/go/errors/errors.yaml +++ b/foreign/go/errors/errors.yaml @@ -777,7 +777,7 @@ type: string - name: TooManyConsumerOffsets code: 3024 - format: "too many consumer offsets" + format: "too many consumer offsets for partition" fields: [] - name: PartitionIdSpaceExhausted code: 3013 diff --git a/foreign/go/errors/errors_gen.go b/foreign/go/errors/errors_gen.go index 67641e984..06031cd66 100644 --- a/foreign/go/errors/errors_gen.go +++ b/foreign/go/errors/errors_gen.go @@ -1586,7 +1586,7 @@ func (e CannotOpenConsumerOffsetsFile) Is(target error) bool { type TooManyConsumerOffsets struct{} -func (e TooManyConsumerOffsets) Error() string { return "too many consumer offsets" } +func (e TooManyConsumerOffsets) Error() string { return "too many consumer offsets for partition" } func (e TooManyConsumerOffsets) Code() Code { return 3024 } func (e TooManyConsumerOffsets) Is(target error) bool { _, ok := target.(TooManyConsumerOffsets) diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.java index 992f3b094..5d776f827 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.java @@ -93,6 +93,8 @@ public enum IggyErrorCode { // Partition errors PARTITION_NOT_FOUND(3007), PARTITION_ID_SPACE_EXHAUSTED(3013), + + // Consumer offset errors TOO_MANY_CONSUMER_OFFSETS(3024), // Segment errors diff --git a/foreign/node/package-lock.json b/foreign/node/package-lock.json index 69b394e60..10bbf5baf 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.5", + "version": "0.10.0-edge.6", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "apache-iggy", - "version": "0.10.0-edge.5", + "version": "0.10.0-edge.6", "license": "Apache-2.0", "dependencies": { "@node-rs/xxhash": "1.7.7", diff --git a/foreign/node/package.json b/foreign/node/package.json index 6722c7161..ea4ec4dd0 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.5", + "version": "0.10.0-edge.6", "description": "Official Apache Iggy NodeJS SDK", "keywords": [ "iggy", diff --git a/foreign/node/src/wire/error.code.test.ts b/foreign/node/src/wire/error.code.test.ts index fc4cb34cb..edbaf7367 100644 --- a/foreign/node/src/wire/error.code.test.ts +++ b/foreign/node/src/wire/error.code.test.ts @@ -20,7 +20,7 @@ import { it } from 'node:test'; import { translateErrorCode } from './error.code.js'; it('translates the consumer-offset capacity error', () => { - assert.equal(translateErrorCode(3024), 'Too many consumer offsets'); + assert.equal(translateErrorCode(3024), 'Too many consumer offsets for partition'); }); it('translates the consumer-group error range', () => { diff --git a/foreign/node/src/wire/error.code.ts b/foreign/node/src/wire/error.code.ts index d0a1f3d78..2cdb31e2f 100644 --- a/foreign/node/src/wire/error.code.ts +++ b/foreign/node/src/wire/error.code.ts @@ -165,7 +165,7 @@ export const translateErrorCode = (code: number): string => { case '3021': return "Consumer offset for consumer with ID: {0} was not found."; case '3022': return "Failed to resolve consumer with ID: {0}"; case '3023': return "Cannot open consumer offsets file for path: {0}"; - case '3024': return "Too many consumer offsets"; + case '3024': return "Too many consumer offsets for partition"; case '3013': return "Partition id space exhausted for this topic"; // MESSAGE diff --git a/foreign/python/Cargo.toml b/foreign/python/Cargo.toml index f3965a16e..1430a98fc 100644 --- a/foreign/python/Cargo.toml +++ b/foreign/python/Cargo.toml @@ -17,7 +17,7 @@ [package] name = "apache-iggy" -version = "0.9.0-dev6" +version = "0.9.0-dev7" edition = "2024" authors = ["Iggy Committers <[email protected]>"] license = "Apache-2.0" @@ -37,7 +37,7 @@ doc = false [dependencies] bytes = "1.12.1" futures = "0.3.34" -iggy = { path = "../../core/sdk", version = "0.11.0-edge.6" } +iggy = { path = "../../core/sdk", version = "0.11.0-edge.7" } paste = "1" pyo3 = "0.29.2" pyo3-async-runtimes = { version = "0.29.0", features = [ diff --git a/foreign/python/pyproject.toml b/foreign/python/pyproject.toml index 54d53ec90..1434010dc 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.dev6" +version = "0.9.0.dev7" 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 d6f3eb506..f4fce79c9 100644 --- a/foreign/python/uv.lock +++ b/foreign/python/uv.lock @@ -8,7 +8,7 @@ exclude-newer-span = "P7D" [[package]] name = "apache-iggy" -version = "0.9.0.dev6" +version = "0.9.0.dev7" source = { editable = "." } [package.optional-dependencies]
