fix(sdk): resume stopped partitions and keep readiness waits current (#4693) A grouped fetch evaluates partitions that another shard owns against a mirror of their state, and the owner pushes commits to that mirror only while a wait parks on it. A wait that another ready partition ended at once answered from the stale mirror, and the fetch read only the partitions that the wait named, so a busy partition on one shard starved a partition with records on another shard of the same node. Such a wait now asks the owners again for the targets that a mirror without pushes found empty, reads a mirror once more when a push refreshed it during the wait, and that mirror then gets the pushes of its partition while waits name it. The reader stops a partition after six lasting refusals in a row and reads it again only after a seek, but the connectors sink never sought, and the C# and Node consumers, whose docs say that the reader retries by itself, never did either. The sink resumes such a partition through a new ReadError::is_lasting_refusal, the same classifier that consume_messages uses. The C#, Node and Swift consumers end with a stop, as consume_messages does, and say which errors stop a partition and which a seek to Next resumes. A stop while a lost committing read is still pending reports that read, in Rust, C# and Node, and the C# and Node readers give up a catch-up gap before every stop, so a seek after the stop skips nothing without a word. A Go probe refusal no longer counts against its group mates. A committing reader started anywhere but Next never read a partition again after a seek to Next, because the walk that resolves a start returned early. It now returns early only when every partition is in a group fetch. with_binary_lineage and with_binary_deadline were async functions that kept the wrapped future twice, which doubled every SDK call that they wrap. Without that, the producer send future shrinks severalfold and its layout gets shallower, with no allocation, so a library can await a send under its own futures without boxing it.
diff --git a/.claude/skills/connector-runtime/SKILL.md b/.claude/skills/connector-runtime/SKILL.md index 66f79e9..e063be0 100644 --- a/.claude/skills/connector-runtime/SKILL.md +++ b/.claude/skills/connector-runtime/SKILL.md
@@ -135,6 +135,8 @@ 2. Spawn one task per topic (`spawn_consume_tasks`). 3. Poll Iggy → batch messages (default `batch_length = 1000`, the poll `count`). Polls use `CommitPolicy::OnRead`: the server stores the offset of each reply, so a reply lost after that store skips its batch. A `consumer.next()` `Err` (a `ReadError`) bumps `errors` and `continue`s. The reader retries most errors by itself. For a partition that it stops, `handle_read_error` seeks: after a read with an unknown outcome (`FetchProgressUncertain`, `PartitionLaneBlocked`) it first calls `abandon_operation` and then seeks `Next` (after the stored offset). It seeks the `skip` of a message that does not decode, and `Next` after `HistoryUnavailable`. + + It also seeks `Next` after six refusals in a row that a retry does not change (`ReadError::is_lasting_refusal`: `Unauthorized`, a missing stream, topic or partition, and the other lasting refusals). A seek to `Next` stores nothing, so it works while the refusal lasts. 4. Decode via `Schema::decoder()`. Decode failure logs + bumps `errors`. `messages_filtered` is reserved for the transform filter contract (`Ok(None)`). 5. Apply transforms (chain). A transform `Err` is logged, bumps `errors`, and drops only that message (drop-and-continue, mirroring the source) - one bad payload never kills the batch. `Ok(None)` filters the message - bumps `messages_filtered{connector_type="sink"}`. 6. Postcard-encode batch as `RawMessages`. Headers as a separate postcard blob. Per-message failures (missing field, payload conversion, header serialization) bump `errors` + skip. Batch-level failures (postcard serialization of metadata / `RawMessages`) propagate `Err` -> `spawn_consume_tasks` wrapper bumps `errors` + `set_error`. The total histogram and benchmark emit still fire for the failed batch before the `Err` propagates.
diff --git a/Cargo.lock b/Cargo.lock index d84261e..ca1dd5e 100644 --- a/Cargo.lock +++ b/Cargo.lock
@@ -7091,7 +7091,7 @@ [[package]] name = "iggy" -version = "0.11.1-edge.8" +version = "0.11.1-edge.9" dependencies = [ "async-broadcast", "async-dropper", @@ -7222,7 +7222,7 @@ [[package]] name = "iggy-connectors" -version = "0.5.1-edge.6" +version = "0.5.1-edge.7" dependencies = [ "apache-avro 0.22.0", "async-trait", @@ -7369,7 +7369,7 @@ [[package]] name = "iggy_common" -version = "0.11.1-edge.9" +version = "0.11.1-edge.10" dependencies = [ "aes-gcm 0.11.1", "aes-gcm-siv", @@ -7389,6 +7389,7 @@ "humantime", "iggy_binary_protocol", "papaya", + "pin-project-lite", "ring", "secrecy", "serde", @@ -12999,7 +13000,7 @@ [[package]] name = "server" -version = "0.9.1-edge.7" +version = "0.9.1-edge.8" dependencies = [ "ahash", "arc-swap",
diff --git a/Cargo.toml b/Cargo.toml index e07b16a..6290018 100644 --- a/Cargo.toml +++ b/Cargo.toml
@@ -214,11 +214,11 @@ iceberg = "0.10.1" iceberg-catalog-rest = "0.10.1" iceberg-storage-opendal = "0.10.1" -iggy = { path = "core/sdk", version = "0.11.1-edge.8" } +iggy = { path = "core/sdk", version = "0.11.1-edge.9" } iggy-cli = { path = "core/cli", version = "0.14.1-edge.7" } iggy-gateway-kafka = { path = "gateways/kafka" } iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.1-edge.8" } -iggy_common = { path = "core/common", version = "0.11.1-edge.9" } +iggy_common = { path = "core/common", version = "0.11.1-edge.10" } iggy_connector_doris_sink = { path = "core/connectors/sinks/doris_sink" } iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.5.1-edge.2" } iggy_kafka_wire = { path = "core/kafka_wire" } @@ -281,6 +281,7 @@ paste = "1.0" percent-encoding = "2.3.2" pgwire = "0.41.1" +pin-project-lite = "0.2.17" postcard = { version = "1.1.3", features = ["alloc"] } predicates = "3.1.4" proc-macro2 = "1"
diff --git a/bdd/python/uv.lock b/bdd/python/uv.lock index e766e0e..d9b15f6 100644 --- a/bdd/python/uv.lock +++ b/bdd/python/uv.lock
@@ -8,7 +8,7 @@ [[package]] name = "apache-iggy" -version = "0.9.1.dev9" +version = "0.9.1.dev10" source = { directory = "../../foreign/python" } [package.metadata]
diff --git a/core/common/Cargo.toml b/core/common/Cargo.toml index 5c267dc..4f3dec8 100644 --- a/core/common/Cargo.toml +++ b/core/common/Cargo.toml
@@ -17,7 +17,7 @@ [package] name = "iggy_common" -version = "0.11.1-edge.9" +version = "0.11.1-edge.10" 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 @@ -46,6 +46,7 @@ humantime = { workspace = true } iggy_binary_protocol = { workspace = true } papaya = { workspace = true } +pin-project-lite = { workspace = true } ring = { workspace = true } secrecy = { workspace = true } serde = { workspace = true }
diff --git a/core/common/src/traits/binary_transport.rs b/core/common/src/traits/binary_transport.rs index 4f88c2a..cb44cda 100644 --- a/core/common/src/traits/binary_transport.rs +++ b/core/common/src/traits/binary_transport.rs
@@ -30,8 +30,12 @@ use iggy_binary_protocol::requests::system::{BindSessionRequest, SessionIdentity}; use iggy_binary_protocol::requests::users::login_register::BindSecret; use iggy_binary_protocol::responses::users::LoginRegisterResponse; +use pin_project_lite::pin_project; +use std::pin::Pin; use std::sync::Arc; +use std::task::{Context, Poll}; use std::time::Duration; +use tokio::task::futures::TaskLocalFuture; use tokio::time::Instant; /// Default budget shared by a binary call's lookups, admission and retries. @@ -68,25 +72,65 @@ } } +pin_project! { + /// [`with_binary_deadline`]: the future until its first poll, then the scope that it runs + /// in, whose deadline needs the deadline in force at that poll. + #[project = DeadlineScopeProjection] + #[project_replace = DeadlineScopeOwned] + enum DeadlineScope<F: Future> { + Waiting { deadline: Instant, future: F }, + Running { #[pin] scope: TaskLocalFuture<Instant, F> }, + /// Only while the first poll moves the future into its scope. + Moving, + } +} + +impl<F: Future> Future for DeadlineScope<F> { + type Output = F::Output; + + fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<F::Output> { + if matches!(*self, Self::Waiting { .. }) + && let DeadlineScopeOwned::Waiting { deadline, future } = + self.as_mut().project_replace(Self::Moving) + { + let deadline = OPERATION_DEADLINE + .try_with(|current| deadline.min(*current)) + .unwrap_or(deadline); + self.set(Self::Running { + scope: OPERATION_DEADLINE.scope(deadline, future), + }); + } + match self.project() { + DeadlineScopeProjection::Running { scope } => scope.poll(context), + DeadlineScopeProjection::Waiting { .. } | DeadlineScopeProjection::Moving => { + Poll::Pending + } + } + } +} + #[doc(hidden)] pub fn binary_operation_lineage() -> Option<BinaryOperationLineage> { OPERATION_LINEAGE.try_get().ok().flatten() } +// Neither helper is an `async fn`: one would keep `future` beside the scope that took it, +// which doubles every call these helpers wrap. #[doc(hidden)] -pub async fn with_binary_lineage<F: Future>( +pub fn with_binary_lineage<F: Future>( lineage: Option<BinaryOperationLineage>, future: F, -) -> F::Output { - OPERATION_LINEAGE.scope(lineage, future).await +) -> impl Future<Output = F::Output> { + OPERATION_LINEAGE.scope(lineage, future) } +/// The deadline is the earlier of `deadline` and the one in force when the future first runs. #[doc(hidden)] -pub async fn with_binary_deadline<F: Future>(deadline: Instant, future: F) -> F::Output { - let deadline = OPERATION_DEADLINE - .try_with(|current| deadline.min(*current)) - .unwrap_or(deadline); - OPERATION_DEADLINE.scope(deadline, future).await +pub fn with_binary_deadline<F: Future>( + deadline: Instant, + future: F, +) -> impl Future<Output = F::Output> { + DeadlineScope::Waiting { deadline, future } } #[doc(hidden)]
diff --git a/core/common/src/types/message/commit_outcome.rs b/core/common/src/types/message/commit_outcome.rs index 74e64bb..ad82b55 100644 --- a/core/common/src/types/message/commit_outcome.rs +++ b/core/common/src/types/message/commit_outcome.rs
@@ -77,6 +77,9 @@ assert!(CommitErrorScope::Section.proves_nothing_stored(&IggyError::TransientNotAccepted)); } + /// The server checks the grant again after the owner's read stored the offset, so a + /// permission lost during a committing read can come back as `Unauthorized` over a batch + /// that is already committed. #[test] fn given_lost_session_when_classifying_then_should_leave_outcome_unknown_in_both_scopes() { for error in [
diff --git a/core/connectors/runtime/Cargo.toml b/core/connectors/runtime/Cargo.toml index fce5d41..447848d 100644 --- a/core/connectors/runtime/Cargo.toml +++ b/core/connectors/runtime/Cargo.toml
@@ -17,7 +17,7 @@ [package] name = "iggy-connectors" -version = "0.5.1-edge.6" +version = "0.5.1-edge.7" description = "Connectors runtime for Iggy message streaming platform" edition = "2024" license = "Apache-2.0"
diff --git a/core/connectors/runtime/src/sink.rs b/core/connectors/runtime/src/sink.rs index 5f2cd38..65ece6b 100644 --- a/core/connectors/runtime/src/sink.rs +++ b/core/connectors/runtime/src/sink.rs
@@ -541,10 +541,15 @@ } /// The reader stops a partition after a read with an unknown outcome, a message that does not -/// decode, or a new partition history. Each would wait for a restart, so the sink resumes the -/// partition. After an unknown outcome it resumes after the stored offset. The abandoned request -/// can still store its offset later, so the sink can skip the messages of that request. The -/// reader retries any other error by itself. +/// decode, a new partition history, or refusals in a row that a quick retry does not change. +/// Each would wait for a restart, so the sink resumes the partition. After an unknown outcome it +/// resumes after the stored offset. The abandoned request can still store its offset later, so +/// the sink can skip the messages of that request. A seek to `Next` stores nothing, so it +/// resumes a refused partition while the refusal lasts. The seek starts a new cycle of the +/// reader, so a partition that stays refused costs six reads with a doubling wait per cycle, +/// about half a minute at the default interval, plus one warning and one error count here. The +/// sink starts at `Next`, so `Next` is where it reads on. The reader retries any other error by +/// itself. async fn handle_read_error( messages: &mut impl SinkMessages, error: ReadError, @@ -587,9 +592,17 @@ ); (partition_id, PollingStrategy::next()) } + (Some(partition_id), _, None) if error.is_lasting_refusal() => { + warn!( + "Partition: {partition_id} of sink connector with ID: {plugin_id} from stream: {stream}, topic: {topic} was stopped by the reader after refusals in a row. It resumes after its stored offset. {}", + error.error + ); + (partition_id, PollingStrategy::next()) + } _ => { error!( - "Failed to receive message for sink connector with ID: {plugin_id} from stream: {stream}, topic: {topic}" + "Failed to receive message for sink connector with ID: {plugin_id} from stream: {stream}, topic: {topic}. {}", + error.error ); return; } @@ -1033,7 +1046,7 @@ mod tests { use super::*; use dashmap::DashMap; - use iggy::prelude::IggyMessageHeader; + use iggy::prelude::{Identifier, IggyMessageHeader}; use iggy_connector_sdk::transforms::TransformType; use iggy_connector_sdk::transforms::{ProtoConvert, ProtoConvertConfig}; use iggy_connector_sdk::{Error, Payload}; @@ -1639,6 +1652,49 @@ } #[tokio::test] + async fn given_lasting_refusals_that_stop_a_partition_when_consumed_should_resume_it() { + let plugin_id = next_plugin_id(); + let metrics = Arc::new(Metrics::init()); + let stream = Identifier::named("test_stream").expect("valid stream name"); + let topic = Identifier::named("test_topic").expect("valid topic name"); + let refusals = [ + IggyError::Unauthorized, + IggyError::ConsumerGroupPartitionNotOwned(1, 2), + IggyError::FeatureUnavailable, + IggyError::InvalidCommand, + IggyError::PartitionNotFound(3, stream.clone(), topic.clone()), + IggyError::StreamIdNotFound(stream.clone()), + IggyError::StreamNameNotFound("test_stream".to_owned()), + IggyError::TopicIdNotFound(topic, stream), + IggyError::TopicNameNotFound("test_topic".to_owned(), "test_stream".to_owned()), + ]; + let count = refusals.len(); + let stops = (1..) + .zip(refusals) + .map(|(partition_id, error)| ReadError { + partition_id: Some(partition_id), + error, + skip: None, + offset: None, + record: None, + }) + .map(Err::<iggy::prelude::ReceivedMessage, _>); + let mut messages = FakeMessages::new(futures::stream::iter(stops)); + + consume_reads(plugin_id, &mut messages, &metrics).await; + + let resumed: Vec<_> = (1..) + .take(count) + .map(|partition_id| Recovery::Seek(partition_id, PollingStrategy::next())) + .collect(); + assert_eq!(messages.calls, resumed); + assert_eq!( + metrics.get_errors("test_sink", ConnectorType::Sink), + count as u64 + ); + } + + #[tokio::test] async fn given_a_read_error_that_the_reader_retries_when_consumed_should_only_count_it() { let plugin_id = next_plugin_id(); let metrics = Arc::new(Metrics::init()); @@ -1657,12 +1713,19 @@ offset: None, record: None, }), + Err(ReadError { + partition_id: None, + error: IggyError::Unauthorized, + skip: None, + offset: None, + record: None, + }), ])); consume_reads(plugin_id, &mut messages, &metrics).await; assert_eq!(messages.calls, []); - assert_eq!(metrics.get_errors("test_sink", ConnectorType::Sink), 2); + assert_eq!(metrics.get_errors("test_sink", ConnectorType::Sink), 3); } #[tokio::test]
diff --git a/core/sdk/Cargo.toml b/core/sdk/Cargo.toml index 62acafe..4870adb 100644 --- a/core/sdk/Cargo.toml +++ b/core/sdk/Cargo.toml
@@ -17,7 +17,7 @@ [package] name = "iggy" -version = "0.11.1-edge.8" +version = "0.11.1-edge.9" 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/producer.rs b/core/sdk/src/clients/producer.rs index 4a3252f..d8e2a71 100644 --- a/core/sdk/src/clients/producer.rs +++ b/core/sdk/src/clients/producer.rs
@@ -2359,6 +2359,21 @@ } } + /// A caller holds the send future in its own state, so a large one costs every caller. + #[test] + fn given_a_direct_producer_when_a_send_is_created_should_hold_at_most_four_kibibytes() { + const MAX_SEND_FUTURE_BYTES: usize = 4096; + let producer = producer_on(IggyRwLock::new(ClientWrapper::Tcp(TcpClient::default()))); + + let send = producer.send_with_partitioning(Vec::new(), None); + + let bytes = std::mem::size_of_val(&send); + assert!( + bytes <= MAX_SEND_FUTURE_BYTES, + "the send future holds {bytes} bytes" + ); + } + #[tokio::test] async fn test_aged_out_request_over_http_stops_producer_retries() { const SEND_REQUEST_LINE: &str = "POST /streams/1/topics/1/messages ";
diff --git a/core/sdk/src/clients/reader.rs b/core/sdk/src/clients/reader.rs index 1695a36..5568319 100644 --- a/core/sdk/src/clients/reader.rs +++ b/core/sdk/src/clients/reader.rs
@@ -42,7 +42,7 @@ use membership::Membership; use progress::Progress; use worker::Worker; -pub(crate) use worker::lasting_refusal; +use worker::lasting_refusal; #[cfg(test)] mod cluster_tests; @@ -622,6 +622,12 @@ record: None, } } + + /// Whether the reader stopped the partition of this error after refusals in a row that a + /// quick retry does not change. Only a [`seek`](PartitionReader::seek) reads it again. + pub fn is_lasting_refusal(&self) -> bool { + self.partition_id.is_some() && self.skip.is_none() && lasting_refusal(&self.error) + } } impl Display for ReadError {
diff --git a/core/sdk/src/clients/reader/cluster_tests.rs b/core/sdk/src/clients/reader/cluster_tests.rs index 5aad8f1..0b119a7 100644 --- a/core/sdk/src/clients/reader/cluster_tests.rs +++ b/core/sdk/src/clients/reader/cluster_tests.rs
@@ -1023,6 +1023,46 @@ ); } +/// A seek to `Next` leaves the partition without a known start, which a poll resolves before a +/// committing fetch reads it, whatever start the reader was built with. +#[tokio::test(start_paused = true)] +async fn given_a_committing_reader_started_at_first_when_a_partition_is_sought_to_next_should_read_it_again() + { + const RESUMED_OFFSET: u64 = 5; + let (client, cluster) = connect_to_three_replicas().await; + script_fetches( + &cluster, + [ + served_answer(PARTITION, 0), + served_answer(PARTITION, RESUMED_OFFSET), + ], + ); + script_initial_next(&cluster, &[(PARTITION, RESUMED_OFFSET)]); + let mut reader = committing(standalone(client, [PARTITION])) + .start(PollingStrategy::first()) + .build() + .await + .unwrap(); + let first = timeout(POLL_TIMEOUT, reader.next()).await.unwrap().unwrap(); + + reader + .seek(PARTITION, PollingStrategy::next()) + .await + .unwrap(); + let resumed = timeout(POLL_TIMEOUT, reader.next()).await; + + assert_eq!(partition_and_offsets(&first), (PARTITION, vec![0])); + let resumed = resumed + .expect("the partition sought to next should be read again") + .unwrap(); + assert_eq!( + partition_and_offsets(&resumed), + (PARTITION, vec![RESUMED_OFFSET]) + ); + assert_eq!(cluster.count(POLL_MESSAGES_DEFERRED_CODE), 1); + reader.close().await.unwrap(); +} + #[tokio::test(start_paused = true)] async fn given_a_partition_refused_beside_a_served_one_when_a_fetch_completes_should_deliver_the_served_batch() {
diff --git a/core/sdk/src/clients/reader/worker.rs b/core/sdk/src/clients/reader/worker.rs index e8d2c2b..d35093b 100644 --- a/core/sdk/src/clients/reader/worker.rs +++ b/core/sdk/src/clients/reader/worker.rs
@@ -947,15 +947,10 @@ let by_group = !self.groups.is_empty(); // Every partition is in a group fetch, so the walk below would find nothing. if by_group - && self.probing.is_empty() - && self.read_alone.is_empty() - && self.catch_up.is_empty() - && !(self.config.on_read() && self.config.start.kind == PollingKind::Next) - && fetched_by_group_kind(self.config.start.kind) && self - .positions - .values() - .all(|position| fetched_by_group_kind(position.kind)) + .partitions + .iter() + .all(|partition| self.fetched_by_group(*partition)) { return; } @@ -1038,13 +1033,14 @@ } fn needs_start(&self, partition: u32) -> bool { + if !self.config.on_read() { + return false; + } let position = self.position(partition); - self.config.on_read() - && matches!( - position.kind, - PollingKind::Next | PollingKind::Last | PollingKind::Timestamp - ) - && position.context().is_none() + matches!( + position.kind, + PollingKind::Next | PollingKind::Last | PollingKind::Timestamp + ) && position.context().is_none() } fn is_idle(&self, partition: u32) -> bool { @@ -2024,6 +2020,8 @@ self.positions.insert(partition, position); } + /// Ends the catch-up of `partition` and reports the range that it did not read again. + /// Returns whether it reported one. fn interrupt_catch_up(&mut self, partition: u32) -> bool { let Some(until) = self.catch_up.remove(&partition) else { return false; @@ -2038,7 +2036,8 @@ .and_then(|retained| retained.batch.messages.first()) .map_or(until, |first| first.header.offset.saturating_sub(1)); let operation = self.lost_operations.remove(&partition).unwrap_or_default(); - if position.value <= lost_end { + let lost = position.value <= lost_end; + if lost { error!( "Partition: {partition} lost offsets from: {} to: {lost_end} during read recovery", position.value @@ -2081,7 +2080,7 @@ }); } } - true + lost } /// Abandons the operations of lost committing reads, so their progress blocks go. The next @@ -2202,7 +2201,7 @@ } IggyError::HistoryUnavailable => { for &partition in partitions { - let interrupted = self.interrupt_catch_up(partition); + let reported = self.interrupt_catch_up(partition); self.shared.progress.forget_undecoded(partition); // A position without a history reads the history of the cached route, which // the driver drops after this refusal, so only a lost history stops. @@ -2216,7 +2215,7 @@ self.positions.remove(&partition); self.retry_at.remove(&partition); self.stopped.insert(partition); - if !interrupted { + if !reported && !self.report_pending_loss(partition) { self.report(ReadError::new(Some(partition), error.clone())); } } @@ -2251,10 +2250,26 @@ self.lasting_refusals.remove(&partition); self.retry_at.remove(&partition); self.stopped.insert(partition); - self.report(ReadError::new(Some(partition), error.clone())); + if !self.interrupt_catch_up(partition) && !self.report_pending_loss(partition) { + self.report(ReadError::new(Some(partition), error.clone())); + } } } + /// A stop before the next read showed what a lost committing read stored leaves that read + /// pending, and a seek to `Next` would then skip its range without a word. So the stop + /// reports the lost read instead of its own error. Returns whether it did. + fn report_pending_loss(&mut self, partition: u32) -> bool { + let Some(&operation) = self.lost_operations.get(&partition) else { + return false; + }; + self.report(ReadError::new( + Some(partition), + IggyError::FetchProgressUncertain(operation), + )); + true + } + /// The driver blocks the partition until `abandon_operation`. With `OnRead`, the partition /// also stops until a seek. fn block(&mut self, partition: u32, operation: u64) { @@ -4815,6 +4830,69 @@ ); } + /// A seek resumes a stopped partition and drops its catch-up, so a lasting stop during a + /// catch-up reports the range that the lost read stored, as the other stops do. + #[tokio::test] + async fn given_a_catch_up_when_lasting_refusals_stop_its_partition_should_report_the_lost_range() + { + let (mut worker, mut receiver) = worker_of(committing(builder_for(group()))); + assert!(worker.apply(grouped(&worker, SESSION, VERSION, vec![vec![0]]))); + worker.positions.insert(0, next_at(0, 5)); + worker.refuse(&[0], IggyError::FetchProgressUncertain(OPERATION)); + worker.complete(polled( + &worker, + FetchKind::Alone, + next_at(0, 5), + 0, + Ok(batch(0, [7, 8], 8)), + )); + assert!( + worker.catch_up.contains_key(&0), + "the skipped offsets are read again" + ); + + for _ in 0..MAX_LASTING_REFUSALS { + worker.refuse(&[0], IggyError::Unauthorized); + } + + assert!(worker.stopped.contains(&0)); + worker.flush_errors(); + let error = next_error(&mut receiver); + assert_eq!(error.error, IggyError::FetchProgressUncertain(OPERATION)); + assert_eq!(error.skip, Some(at(0, 7))); + assert_eq!(offsets(&next_batches(&mut receiver)[0]), [7, 8]); + assert!(worker.catch_up.is_empty()); + } + + /// A lost committing read is pending until the next read shows what it stored. A stop before + /// that read would leave a seek to `Next` to skip the range without a word, so the stop + /// reports the lost read instead. + #[tokio::test] + async fn given_a_pending_lost_read_when_lasting_refusals_stop_its_partition_should_report_the_lost_read() + { + let (mut worker, mut receiver) = worker_of(committing(builder_for(group()))); + assert!(worker.apply(grouped(&worker, SESSION, VERSION, vec![vec![0]]))); + worker.positions.insert(0, next_at(0, 5)); + worker.refuse(&[0], IggyError::FetchProgressUncertain(OPERATION)); + assert!(worker.lost_operations.contains_key(&0)); + + for _ in 0..MAX_LASTING_REFUSALS { + worker.refuse(&[0], IggyError::Unauthorized); + } + + assert!(worker.stopped.contains(&0)); + worker.flush_errors(); + let error = next_error(&mut receiver); + assert_eq!( + (error.partition_id, error.error, error.skip), + (Some(0), IggyError::FetchProgressUncertain(OPERATION), None) + ); + assert!( + worker.lost_operations.contains_key(&0), + "the lost read stays pending until a seek takes the partition" + ); + } + /// A member that the server keeps refusing after each refresh loops without a sign, so the /// refusal counts as lasting, while each one still drops the position and refreshes. #[tokio::test(start_paused = true)]
diff --git a/core/sdk/src/consumer_ext/consumer_message_ext.rs b/core/sdk/src/consumer_ext/consumer_message_ext.rs index ee7415e..b74fe11 100644 --- a/core/sdk/src/consumer_ext/consumer_message_ext.rs +++ b/core/sdk/src/consumer_ext/consumer_message_ext.rs
@@ -16,7 +16,6 @@ // under the License. use crate::clients::consumer::IggyConsumer; -use crate::clients::reader::lasting_refusal; use crate::consumer_ext::{IggyConsumerMessageExt, MessageConsumer}; use crate::prelude::{CommitAfter, IggyError, ReadError}; use async_trait::async_trait; @@ -150,7 +149,7 @@ | IggyError::InvalidMessagesCount | IggyError::ClientShutdown ) - || (err.partition_id.is_some() && lasting_refusal(&err.error)) + || err.is_lasting_refusal() || !self.client_can_recover().await; if ends { error!("Client error: {err} for consumer: {name} on topic: {topic} and stream: {stream}",
diff --git a/core/sdk/tests/send_future_depth.rs b/core/sdk/tests/send_future_depth.rs new file mode 100644 index 0000000..22299c0 --- /dev/null +++ b/core/sdk/tests/send_future_depth.rs
@@ -0,0 +1,78 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! A library awaits a producer send inside futures of its own, and its users await the library +//! inside theirs. The compiler computes the layout of each future through every future it awaits +//! and fails past the crate's recursion limit, 128 by default. This target builds sixteen nested +//! layers over a send under a limit well below the default, so a deeper send future fails its +//! build. + +#![recursion_limit = "72"] + +use iggy::prelude::{ + DirectConfig, IggyClient, IggyError, IggyProducer, Partitioning, SendMessagesResponse, +}; + +async fn layer_0(producer: &IggyProducer) -> Result<SendMessagesResponse, IggyError> { + producer.send_with_partitioning(Vec::new(), None).await +} + +macro_rules! layers { + ($($outer:ident => $inner:ident),+ $(,)?) => { + $( + async fn $outer(producer: &IggyProducer) -> Result<SendMessagesResponse, IggyError> { + $inner(producer).await + } + )+ + }; +} + +layers!( + layer_1 => layer_0, + layer_2 => layer_1, + layer_3 => layer_2, + layer_4 => layer_3, + layer_5 => layer_4, + layer_6 => layer_5, + layer_7 => layer_6, + layer_8 => layer_7, + layer_9 => layer_8, + layer_10 => layer_9, + layer_11 => layer_10, + layer_12 => layer_11, + layer_13 => layer_12, + layer_14 => layer_13, + layer_15 => layer_14, + layer_16 => layer_15, +); + +/// The size of the future that `create` returns, which makes the compiler compute its layout. +fn layout_bytes<F: Future>(_create: impl FnOnce() -> F) -> usize { + size_of::<F>() +} + +#[test] +fn given_sixteen_layers_over_a_send_when_built_under_a_recursion_limit_of_72_should_compile() { + let producer = IggyClient::default() + .producer("stream", "topic") + .expect("valid names") + .partitioning(Partitioning::partition_id(0)) + .direct(DirectConfig::builder().build()) + .build(); + + assert!(layout_bytes(|| layer_16(&producer)) > 0); +}
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml index c4e5841..9e6f8f9 100644 --- a/core/server/Cargo.toml +++ b/core/server/Cargo.toml
@@ -17,7 +17,7 @@ [package] name = "server" -version = "0.9.1-edge.7" +version = "0.9.1-edge.8" edition = "2024" license = "Apache-2.0" publish = false
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 131be88..c651f29 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs
@@ -37,6 +37,7 @@ OversizedTurn, PollQuotaIdentity, ReadinessTarget, ReadinessWait, ReadinessWatch, ReadyTarget, ServedState, WatchedState, }; +use poll::deferred::{MirrorSince, MirrorUse}; pub use poll::{ConsumerAttachment, PollCompleted}; /// How often the replication gauges and partition reports refresh. pub use primary_rebalance::PROBE_INTERVAL as PARTITION_REPORT_INTERVAL; @@ -2758,7 +2759,12 @@ /// adds its own targets that hold messages, also below `min_count`. An /// empty list means that the wait expired without messages. An owner /// that gets an `Offset` target starts to push its state here, so that - /// the next waits on it stay on this shard. + /// the next waits on it stay on this shard. A mirror without those pushes + /// can lag its partition, so a wait that ends on other targets asks the + /// owners again for the targets that such a mirror found empty, unless a + /// push refreshed the mirror during the wait, in which case the wait reads + /// the mirror once more and asks the owner only when the mirror still + /// holds nothing and its owner stopped pushing. /// /// # Errors /// Returns the first rejection. A target that no shard owns, or an inbox @@ -2787,6 +2793,7 @@ let mirrors = !wait.is_zero(); let mut evaluators: BTreeMap<u16, (Vec<ReadinessTarget>, Vec<IggyNamespace>)> = BTreeMap::new(); + let mut snapshots = Vec::new(); for target in targets { let owner = self .shards_table @@ -2794,10 +2801,18 @@ .ok_or(IggyError::TransientNotAccepted)?; let mirrorable = mirrors && owner != self.id && target.strategy.kind == PollingKind::Offset; - let shard = if mirrorable && self.mirror_serves(owner, &target, context.primary, now) { - self.id + let mirror = if mirrorable { + self.mirror_use(owner, &target, context.primary, now) } else { - owner + MirrorUse::Owner + }; + let shard = match mirror { + MirrorUse::Owner => owner, + MirrorUse::Current => self.id, + MirrorUse::Snapshot { pushes } => { + snapshots.push((target.clone(), pushes)); + self.id + } }; let (targets, watched) = evaluators.entry(shard).or_default(); if mirrorable && shard == owner { @@ -2896,6 +2911,44 @@ if let Some(waiter) = parked.waiters.remove(&self.id) { ready.extend(self.take_readiness_wait(waiter)?); } + // A push during the wait can land after the last evaluation of the + // wait, so a mirror that one refreshed is read once more. One that + // still holds nothing is trusted only while its owner keeps pushing. + let mut stale = Vec::new(); + for (target, pushes) in snapshots { + if ready + .iter() + .any(|ready| ready.partition_id == target.partition_id) + { + continue; + } + let since = self.mirror_since(target.namespace, pushes); + if since != MirrorSince::Unchanged + && let Some(found) = self.mirror_ready(context.primary, &target) + { + ready.push(found); + } else if since != MirrorSince::Current { + stale.push(target); + } + } + if !ready.is_empty() && !stale.is_empty() { + let namespaces = stale.iter().map(|target| target.namespace).collect(); + let budget = std::time::Duration::from_micros( + deadline.saturating_sub(self.bus.monotonic_micros()), + ); + // A zero wait asks the owners and never a mirror. An owner that + // cannot answer within the wait leaves the answer as it is. + let owners = Box::pin(self.wait_for_messages( + context, + stale, + std::time::Duration::ZERO, + ANY_MESSAGE, + )); + if let Some(Ok(owned)) = bus_timeout(&self.bus, budget, owners).await { + ready.extend(owned); + } + self.hold_mirrors(namespaces); + } Ok(ready) }
diff --git a/core/shard/src/poll/deferred.rs b/core/shard/src/poll/deferred.rs index c2b1adf..8aa40fc 100644 --- a/core/shard/src/poll/deferred.rs +++ b/core/shard/src/poll/deferred.rs
@@ -240,7 +240,7 @@ } /// One partition of a readiness wait, resolved and authorized by the origin. -#[derive(Debug)] +#[derive(Debug, Clone)] pub struct ReadinessTarget { pub namespace: IggyNamespace, pub partition_id: u32, @@ -640,9 +640,39 @@ used: u64, /// When this shard applied the last push. pushed: u64, + /// The pushes that this shard applied, so that a wait tells whether one + /// came after it started. + pushes: u64, /// The owner took the news that this shard holds a parked wait on the /// partition, so its commits come as pushes. interest: bool, + /// A wait found this mirror behind its partition without a parked wait + /// to refresh it, so the pushes go on while waits name the mirror. + held: bool, +} + +/// What a mirror received since a wait read it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MirrorSince { + /// No push. + Unchanged, + /// A push, and the owner pushes each commit here, so the mirror is as + /// current as one that a parked wait refreshed. + Current, + /// A push that ended the pushes, because it found no parked wait. + Pushed, +} + +/// How a wait evaluates a target that another shard owns. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MirrorUse { + /// The owner evaluates it. + Owner, + /// The mirror evaluates it, and the owner pushes each commit here. + Current, + /// The mirror evaluates it from its last push, the `pushes`-th, which can + /// miss newer records. + Snapshot { pushes: u64 }, } impl Mirror { @@ -2282,7 +2312,7 @@ .entry(mirror.owner) .or_default() .push((namespace, mirror.seq)); - } else if parked && !mirror.interest { + } else if (parked || mirror.held) && !mirror.interest { // An owner whose inbox refused the news hears it again. interested.push(namespace); } @@ -2370,12 +2400,13 @@ // the next wait can still read the mirror. mirror.state = update.state; mirror.pushed = now; + mirror.pushes = mirror.pushes.wrapping_add(1); let parked = state.partitions.contains_key(&update.namespace); if mirror.unused(now) && !parked { unused.push((update.namespace, update.seq)); continue; } - if !parked && mirror.interest { + if !parked && mirror.interest && !mirror.held { idle.push(update.namespace); } signals.mark(update.namespace); @@ -2467,27 +2498,84 @@ } } - /// Whether this shard evaluates `target` against its mirror. A mirror - /// that finds an error in the partition, or that lags the install of a - /// group owner, sends the target to the owner, so that no wait gets a - /// stale error from a mirror. - pub(crate) fn mirror_serves( + /// How this shard evaluates `target` of `owner`. A mirror that finds an + /// error in the partition, or that lags the install of a group owner, + /// sends the target to the owner, so that no wait gets a stale error from + /// a mirror. + pub(crate) fn mirror_use( &self, owner: u16, target: &ReadinessTarget, primary: bool, now: u64, - ) -> bool { + ) -> MirrorUse { let mut mirrors = self.deferred_polls.mirrors.borrow_mut(); let Some(mirror) = mirrors.entries.get_mut(&target.namespace) else { - return false; + return MirrorUse::Owner; }; mirror.used = now; - mirror.owner == owner + let serves = mirror.owner == owner && mirror .state .as_ref() - .is_some_and(|state| matches!(state.check(primary, target), Ok(true))) + .is_some_and(|state| matches!(state.check(primary, target), Ok(true))); + if !serves { + MirrorUse::Owner + } else if mirror.interest { + MirrorUse::Current + } else { + MirrorUse::Snapshot { + pushes: mirror.pushes, + } + } + } + + /// How the mirror of `namespace` changed since a wait read it after its + /// `pushes`-th push. + pub(crate) fn mirror_since(&self, namespace: IggyNamespace, pushes: u64) -> MirrorSince { + match self.deferred_polls.mirrors.borrow().entries.get(&namespace) { + Some(mirror) if mirror.pushes == pushes => MirrorSince::Unchanged, + Some(mirror) if mirror.interest => MirrorSince::Current, + Some(_) => MirrorSince::Pushed, + None => MirrorSince::Unchanged, + } + } + + /// `target` as its mirror finds it now, if it holds messages. A wait reads + /// a mirror this way after the pushes that came while it waited, which no + /// evaluation of the wait saw. + pub(crate) fn mirror_ready( + &self, + primary: bool, + target: &ReadinessTarget, + ) -> Option<ReadyTarget> + where + M: StreamsFrontend, + { + let readiness = self.evaluate_readiness(primary, target).ok()?; + (readiness.available > 0).then_some(ReadyTarget { + partition_id: target.partition_id, + available: readiness.available, + known_bytes: None, + }) + } + + /// Asks the owners of the mirrors of `namespaces` to push each commit + /// here until no wait names the mirror, so the next waits on them need + /// no second evaluation. + pub(crate) fn hold_mirrors(&self, namespaces: Vec<IggyNamespace>) + where + M: StreamsFrontend, + { + { + let mut mirrors = self.deferred_polls.mirrors.borrow_mut(); + for namespace in &namespaces { + if let Some(mirror) = mirrors.entries.get_mut(namespace) { + mirror.held = true; + } + } + } + self.tell_mirror_owners(namespaces, true); } /// Mirrors `namespaces` under `seq`, after `owner` took the frame that @@ -2508,11 +2596,14 @@ state: None, used: now, pushed: now, + pushes: 0, interest: false, + held: false, }); // A new owner holds no news of this shard's parked waits. if mirror.owner != owner { mirror.interest = false; + mirror.held = false; } mirror.owner = owner; mirror.seq = seq;
diff --git a/core/shard/src/poll/deferred_tests.rs b/core/shard/src/poll/deferred_tests.rs index d22e27c..ee825e1 100644 --- a/core/shard/src/poll/deferred_tests.rs +++ b/core/shard/src/poll/deferred_tests.rs
@@ -2097,6 +2097,17 @@ shard.flush_readiness_pushes(); } +/// Applies the next push of `inbox` alone, without the service loop that +/// [`pump`] runs after every frame. +fn apply_next_push(shard: &TestShard, inbox: &Inbox) { + let Ok(ShardFrame::Lifecycle(LifecycleFrame::ReadinessUpdate { from, states })) = + inbox.replies.try_recv().or_else(|_| inbox.main.try_recv()) + else { + panic!("a push is next in the inbox"); + }; + shard.apply_readiness_update(from, states); +} + /// A first wait for `partition` goes to the owner. The owner answers it and /// pushes the state, which the watcher then mirrors. async fn mirror(watcher: &TestShard, owner: &TestShard, inboxes: &[Inbox], partition: usize) { @@ -2305,6 +2316,183 @@ } #[compio::test] +async fn wait_that_ends_on_another_ready_target_asks_the_owner_of_a_mirror_without_pushes() { + let (watcher, owner, inboxes) = watched_pair(&[&["first"], &["first"]]).await; + mirror(&watcher, &owner, &inboxes, 1).await; + // No wait parks on the mirror, so the owner pushes nothing of this commit. + commit_to(&owner, 1, &["second"], 2).await; + pump(&owner, &inboxes[1]).await; + assert!(inboxes[0].replies.try_recv().is_err()); + let targets = vec![ + readiness_target(&watcher, 0, PollingStrategy::offset(0)), + readiness_target(&watcher, 1, PollingStrategy::offset(1)), + ]; + + let (ready, ()) = futures::join!( + watcher.wait_for_messages( + readiness_caller(), + targets, + PARTITION_READ_TIMEOUT, + ANY_MESSAGE + ), + async { + pump(&owner, &inboxes[1]).await; + yield_now().await; + pump(&owner, &inboxes[1]).await; + } + ); + + assert_eq!( + ids(&ready.unwrap()), + [0, 1], + "the answer names the mirrored partition that holds a record" + ); +} + +#[compio::test] +async fn mirror_that_a_wait_found_behind_keeps_its_pushes_while_waits_name_it() { + let (watcher, owner, inboxes) = watched_pair(&[&["first"], &["first"]]).await; + mirror(&watcher, &owner, &inboxes, 1).await; + commit_to(&owner, 1, &["second"], 2).await; + pump(&owner, &inboxes[1]).await; + let behind = vec![ + readiness_target(&watcher, 0, PollingStrategy::offset(0)), + readiness_target(&watcher, 1, PollingStrategy::offset(1)), + ]; + let (ready, ()) = futures::join!( + watcher.wait_for_messages( + readiness_caller(), + behind, + PARTITION_READ_TIMEOUT, + ANY_MESSAGE + ), + async { + pump(&owner, &inboxes[1]).await; + yield_now().await; + pump(&owner, &inboxes[1]).await; + } + ); + assert_eq!(ids(&ready.unwrap()), [0, 1]); + pump(&owner, &inboxes[1]).await; + pump(&watcher, &inboxes[0]).await; + // Without a parked wait, only the hold keeps these commits coming. + for request in [3, 4] { + commit_to(&owner, 1, &["more"], request).await; + pump(&owner, &inboxes[1]).await; + pump(&watcher, &inboxes[0]).await; + } + let current = vec![ + readiness_target(&watcher, 0, PollingStrategy::offset(0)), + readiness_target(&watcher, 1, PollingStrategy::offset(3)), + ]; + + let (ready, ()) = futures::join!( + watcher.wait_for_messages( + readiness_caller(), + current, + PARTITION_READ_TIMEOUT, + ANY_MESSAGE + ), + async { + yield_now().await; + assert!( + inboxes[1].main.try_recv().is_err(), + "the wait asks the owner nothing" + ); + pump(&owner, &inboxes[1]).await; + } + ); + + assert_eq!(ids(&ready.unwrap()), [0, 1]); +} + +#[compio::test] +async fn wait_whose_owner_refuses_the_second_evaluation_keeps_its_answer() { + let (watcher, owner, inboxes) = watched_pair(&[&["first"], &["first"]]).await; + mirror(&watcher, &owner, &inboxes, 1).await; + commit_to(&owner, 1, &["second"], 2).await; + pump(&owner, &inboxes[1]).await; + let targets = vec![ + readiness_target(&watcher, 0, PollingStrategy::offset(0)), + readiness_target(&watcher, 1, PollingStrategy::offset(1)), + ]; + + let (ready, ()) = futures::join!( + watcher.wait_for_messages( + readiness_caller(), + targets, + PARTITION_READ_TIMEOUT, + ANY_MESSAGE + ), + async { + pump(&owner, &inboxes[1]).await; + // A full inbox refuses the frame that asks the owner again. + while watcher.senders[usize::from(OWNER)] + .try_send(ShardFrame::lifecycle(LifecycleFrame::ReadinessUnwatch { + from: WATCHER, + watches: Vec::new(), + })) + .is_ok() + {} + yield_now().await; + } + ); + + assert_eq!( + ids(&ready.unwrap()), + [0], + "the answer keeps the target that was ready" + ); +} + +#[compio::test] +async fn wait_reads_a_mirror_that_a_push_refreshed_after_its_last_evaluation() { + let (watcher, owner, inboxes) = watched_pair(&[&["first"], &["first"]]).await; + mirror(&watcher, &owner, &inboxes, 0).await; + mirror(&watcher, &owner, &inboxes, 1).await; + let targets = (0..2) + .map(|partition| readiness_target(&watcher, partition, PollingStrategy::offset(1))) + .collect(); + + let (ready, ()) = futures::join!( + watcher.wait_for_messages( + readiness_caller(), + targets, + PARTITION_READ_TIMEOUT, + ANY_MESSAGE + ), + async { + // The park asks for pushes, and each commit then comes in a frame of its own. + pump(&owner, &inboxes[1]).await; + pump(&watcher, &inboxes[0]).await; + commit_to(&owner, 0, &["second"], 2).await; + pump(&owner, &inboxes[1]).await; + commit_to(&owner, 1, &["second"], 3).await; + pump(&owner, &inboxes[1]).await; + // The first push fires the wait, and the second lands before the wait reads its answer. + apply_next_push(&watcher, &inboxes[0]); + watcher.service_deferred_polls().await; + apply_next_push(&watcher, &inboxes[0]); + } + ); + + assert_eq!( + ids(&ready.unwrap()), + [0, 1], + "the answer names the partition whose push came after the wait fired" + ); + while let Ok(frame) = inboxes[1].main.try_recv() { + assert!( + !matches!( + frame, + ShardFrame::Lifecycle(LifecycleFrame::ReadinessWait { .. }) + ), + "the wait asks the owner nothing" + ); + } +} + +#[compio::test] async fn unused_mirror_ends_its_watch() { for commits in [true, false] { let (watcher, owner, inboxes) = watched_pair(&[&["first"]]).await;
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 650ceec..3244419 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs
@@ -2258,6 +2258,9 @@ mod replication_flow_tests; #[cfg(test)] +mod readiness_mirror_tests; + +#[cfg(test)] mod tests { use super::partition_repair_driver_tests::{ cluster, is_commit_for, prepare_for, sharded_cluster,
diff --git a/core/simulator/src/readiness_mirror_tests.rs b/core/simulator/src/readiness_mirror_tests.rs new file mode 100644 index 0000000..942c98e --- /dev/null +++ b/core/simulator/src/readiness_mirror_tests.rs
@@ -0,0 +1,166 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! A grouped fetch (109) evaluates a partition that another shard of the same +//! replica owns against a mirror of its state on the shard of the fetch's +//! connection. The owner pushes each commit to that mirror only while a wait +//! parks on it, so a fetch that another ready partition ends at once must not +//! answer from a mirror that missed newer records. Each round commits one +//! record to each of two partitions that different shards own, then fetches +//! both, and every round must return both records, whichever shard the fetch +//! enters. + +use bytes::Bytes; +use consensus::MetadataHandle; +use iggy_binary_protocol::responses::messages::fetch_messages::FetchMessagesResponse; +use iggy_binary_protocol::{Operation, ReplyHeader, RoutedRequestHeader, WirePollingStrategy}; +use server_common::Message; +use server_common::sharding::IggyNamespace; +use shard::shards_table::calculate_shard_assignment; + +use crate::Simulator; +use crate::client::SimClient; +use crate::packet::PacketSimulatorOptions; + +const CLIENT: u128 = 1; +/// Leads view 0. +const PRIMARY: u8 = 0; +const REPLICAS: u8 = 3; +const SEED: u64 = 0x4D49_5252; +const SHARDS: u16 = 2; +const ROUNDS: u64 = 8; +const REPLY_STEPS: usize = 400; +const FETCH_COUNT: u32 = 16; +/// Partition ids to look through for one that the other shard owns. +const PARTITION_CANDIDATES: usize = 16; +/// Far beyond a round, so a fetch that waits instead of answering at once +/// returns no reply within `REPLY_STEPS`. +const WAIT_US: u64 = 60_000_000; +const TIMEOUT_US: u64 = 61_000_000; + +#[test] +fn given_records_on_partitions_of_two_shards_when_fetched_each_round_should_return_every_record() { + let (mut sim, client, partitions) = cluster(); + let ids: Vec<u32> = partitions.into_iter().map(partition_id).collect(); + for round in 0..ROUNDS { + for namespace in partitions { + produce(&mut sim, &client, namespace, round); + } + let request = client.fetch_messages( + &partitions, + &WirePollingStrategy::offset(round), + (FETCH_COUNT, 1), + false, + (WAIT_US, TIMEOUT_US), + ); + let reply = submit_and_await(&mut sim, &client, request); + + assert_eq!( + reply.header().status, + 0, + "round {round}: the fetch is served" + ); + let response = FetchMessagesResponse::decode(reply.body()).expect("a fetch reply decodes"); + let mut served: Vec<u32> = response + .sections + .iter() + .filter(|section| section.header.messages_count == 1) + .map(|section| section.header.partition_id) + .collect(); + served.sort_unstable(); + assert_eq!( + served, ids, + "round {round}: the fetch returns the new record of each partition" + ); + } +} + +/// Three replicas with two shards each, and the first two partitions of one topic +/// that different shards own. +fn cluster() -> (Simulator, SimClient, [IggyNamespace; 2]) { + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { + enabled: false, + size: iggy_common::IggyByteSize::from(0u64), + bucket_capacity: 1, + }); + let network = PacketSimulatorOptions { + node_count: REPLICAS, + client_count: 1, + seed: SEED, + ..PacketSimulatorOptions::default() + }; + let mut sim = Simulator::with_shards_shell( + usize::from(REPLICAS), + SHARDS, + std::iter::once(CLIENT), + network, + ); + let first = IggyNamespace::new(0, 0, 0); + let owner = + |namespace: &IggyNamespace| calculate_shard_assignment(namespace, u32::from(SHARDS)); + let second = (1..PARTITION_CANDIDATES) + .map(|partition| IggyNamespace::new(0, 0, partition)) + .find(|namespace| owner(namespace) != owner(&first)) + .expect("some partition hashes to the other shard"); + for namespace in [first, second] { + sim.init_partition(namespace); + sim.seed_stream_topic_partition(namespace); + } + for replica in &sim.replicas { + replica.shards[0] + .plane + .metadata() + .seed_applied_frontier_from_consensus(); + } + let client = SimClient::new(CLIENT); + sim.shell_login(&client); + sim.discover_partition_contexts(&client); + (sim, client, [first, second]) +} + +fn produce(sim: &mut Simulator, client: &SimClient, namespace: IggyNamespace, round: u64) { + let payload = Bytes::from(format!("record {round}")); + let reply = sim.await_setup_reply( + client, + PRIMARY, + &client.send_messages(namespace, &[payload]).into_generic(), + Operation::SendMessages, + "produce", + ); + assert_eq!(reply.header().status, 0, "the produce commits"); +} + +fn submit_and_await( + sim: &mut Simulator, + client: &SimClient, + request: Message<RoutedRequestHeader>, +) -> Message<ReplyHeader> { + let request_id = request.header().request; + sim.submit_request(client.client_id(), PRIMARY, request.into_generic()); + for _ in 0..REPLY_STEPS { + if let Some(reply) = sim.step().into_iter().find(|reply| { + reply.header().client == client.client_id() && reply.header().request == request_id + }) { + return reply; + } + } + panic!("request {request_id} got no reply within {REPLY_STEPS} steps"); +} + +fn partition_id(namespace: IggyNamespace) -> u32 { + u32::try_from(namespace.partition_id()).expect("partition id fits u32") +}
diff --git a/examples/node/package-lock.json b/examples/node/package-lock.json index 0e0c239..8fbe10b 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.1-edge.8", + "version": "0.10.1-edge.9", "license": "Apache-2.0", "dependencies": { "@node-rs/xxhash": "1.7.8",
diff --git a/examples/python/uv.lock b/examples/python/uv.lock index f1c66f9..eb5bf71 100644 --- a/examples/python/uv.lock +++ b/examples/python/uv.lock
@@ -8,7 +8,7 @@ [[package]] name = "apache-iggy" -version = "0.9.1.dev9" +version = "0.9.1.dev10" source = { directory = "../../foreign/python" } [package.metadata]
diff --git a/foreign/cpp/Cargo.toml b/foreign/cpp/Cargo.toml index bdcf5d2..3bac59c 100644 --- a/foreign/cpp/Cargo.toml +++ b/foreign/cpp/Cargo.toml
@@ -17,7 +17,7 @@ [package] name = "iggy-cpp" -version = "0.1.10" +version = "0.1.11" edition = "2024" license = "Apache-2.0"
diff --git a/foreign/cpp/MODULE.bazel b/foreign/cpp/MODULE.bazel index 9ea7c4d..4ad6286 100644 --- a/foreign/cpp/MODULE.bazel +++ b/foreign/cpp/MODULE.bazel
@@ -17,7 +17,7 @@ module( name = "iggy_cpp", - version = "0.1.10", + version = "0.1.11", ) bazel_dep(name = "rules_cc", version = "0.2.26")
diff --git a/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs b/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs index 7d9c336..7888115 100644 --- a/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs +++ b/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs
@@ -44,8 +44,11 @@ /// before a rebalance takes a partition away, and on shutdown. /// </para> /// <para> -/// A read error does not end the stream: the reader retries by itself, and the error reaches the callbacks of -/// <see cref="SubscribeToErrorEvents" />. +/// A read error that the reader retries by itself does not end the stream, it reaches the callbacks of +/// <see cref="SubscribeToErrorEvents" />. An error that stops a partition (<see cref="ReadException.StopsPartition" />) +/// ends the stream with its <see cref="ReadException" />, as a message that does not decode does, and the +/// partition reads nothing until <see cref="SeekAsync" />. <see cref="ReadException.IsResumable" /> says that a +/// seek to <see cref="PollingStrategy.Next" /> reads it again after its stored offset. /// </para> /// </remarks> public partial class IggyConsumer : IAsyncDisposable @@ -62,6 +65,7 @@ private readonly ILogger<IggyConsumer> _logger; private readonly ILoggerFactory _loggerFactory; private readonly CancellationTokenSource _lifetime = new(); + private TaskCompletionSource? _handling; private TaskCompletionSource? _shutdown; private PartitionReader? _reader; @@ -148,6 +152,11 @@ builder.WithPartitions([.. _config.Partitions]); } + if (_config.RetryInterval is { } retryInterval) + { + builder.WithRetryInterval(retryInterval); + } + _reader = await builder.BuildAsync(ct); Volatile.Write(ref _state, StateReading); } @@ -374,8 +383,9 @@ /// <summary> /// The next batch of the reader, or null once the consumer shut down or <paramref name="ct" /> was canceled. A - /// read error goes to the error events and the reader goes on. A message that does not decode ends the stream - /// with its <see cref="ReadException" />, whose <see cref="ReadException.Skip" /> passes it. + /// read error that the reader retries goes to the error events and the reader goes on. A message that does + /// not decode, whose <see cref="ReadException.Skip" /> passes it, and an error that stops a partition end the + /// stream with their <see cref="ReadException" />. /// </summary> private protected async Task<PolledMessages?> NextBatchAsync(CancellationToken ct) { @@ -386,7 +396,7 @@ { return await Reader.NextAsync(cancellation.Token); } - catch (ReadException error) when (error.Skip is null) + catch (ReadException error) when (error.Skip is null && !error.StopsPartition) { _consumerErrorEvents.Publish(new ConsumerErrorEventArgs(error, error.Message)); }
diff --git a/foreign/csharp/Iggy_SDK/Consumers/IggyConsumerConfig.cs b/foreign/csharp/Iggy_SDK/Consumers/IggyConsumerConfig.cs index b9d0232..70219bc 100644 --- a/foreign/csharp/Iggy_SDK/Consumers/IggyConsumerConfig.cs +++ b/foreign/csharp/Iggy_SDK/Consumers/IggyConsumerConfig.cs
@@ -152,4 +152,7 @@ /// See <see cref="IggyClientConfigurator.HeartbeatInterval" />. /// </summary> public TimeSpan HeartbeatInterval { get; set; } = TimeSpan.FromSeconds(5); + + /// <summary>The wait before a refused partition reads again. Null keeps the default of the reader.</summary> + internal TimeSpan? RetryInterval { get; set; } }
diff --git a/foreign/csharp/Iggy_SDK/Consumers/PartitionReader.cs b/foreign/csharp/Iggy_SDK/Consumers/PartitionReader.cs index 3e759ac..bd39ff2 100644 --- a/foreign/csharp/Iggy_SDK/Consumers/PartitionReader.cs +++ b/foreign/csharp/Iggy_SDK/Consumers/PartitionReader.cs
@@ -1496,7 +1496,7 @@ state.Lasting++; if (state.Lasting >= MaxLastingRefusals) { - report = new ReadException(target.Partition, error); + report = StopReport(target.Partition, state, error, true); state.Stopped = report; break; } @@ -1510,15 +1510,18 @@ target.Partition); state.RetryAt = now + interval; return; - case IggyInvalidStatusCodeException { StatusCode: VsrError.HISTORY_UNAVAILABLE or VsrError.INVALID_SIZE_BYTES }: - report = new ReadException(target.Partition, error); + case IggyInvalidStatusCodeException { StatusCode: VsrError.HISTORY_UNAVAILABLE or VsrError.INVALID_SIZE_BYTES } refusal: + GiveUpGap(target.Partition, state); + report = StopReport(target.Partition, state, error, + refusal.StatusCode == VsrError.HISTORY_UNAVAILABLE); state.Stopped = report; break; case IggyInvalidStatusCodeException status when IsLasting(status.StatusCode): state.Lasting++; if (state.Lasting >= MaxLastingRefusals) { - report = new ReadException(target.Partition, error); + GiveUpGap(target.Partition, state); + report = StopReport(target.Partition, state, error, true); state.Stopped = report; break; } @@ -1550,6 +1553,21 @@ } } + /// <summary> + /// The stop of a partition. A committing read that lost its reply stays pending until the next read shows what + /// it stored, and a seek to Next after the stop would skip that range without a word, so the stop reports the + /// lost read instead of its own error. + /// </summary> + private static ReadException StopReport(uint partition, PartitionState state, Exception error, bool resumable) + { + var reported = state.Lost + ? VsrError.Exception(VsrError.FETCH_PROGRESS_UNCERTAIN, + $"A lost read may have stored offsets of partition {partition} from {state.Position.Value}, and the " + + $"partition stopped on: {error.Message}") + : error; + return new ReadException(partition, reported) { StopsPartition = true, IsResumable = resumable }; + } + private void GiveUpGap(uint partition, PartitionState state) { if (state.CatchUpUntil is not { } until)
diff --git a/foreign/csharp/Iggy_SDK/Consumers/ReadException.cs b/foreign/csharp/Iggy_SDK/Consumers/ReadException.cs index 1fa8295..21c0972 100644 --- a/foreign/csharp/Iggy_SDK/Consumers/ReadException.cs +++ b/foreign/csharp/Iggy_SDK/Consumers/ReadException.cs
@@ -40,6 +40,20 @@ public PollingStrategy? Skip { get; } = skip; /// <summary> + /// The reader stopped the partition of this error until a <see cref="PartitionReader.SeekAsync" />. Every + /// <see cref="IsResumable" /> error stops it, and so does status 80 (InvalidSizeBytes) for a message above the + /// byte limit of a read. + /// </summary> + public bool StopsPartition { get; internal init; } + + /// <summary> + /// The reader stopped the partition after six refusals in a row that a quick retry does not change, such as + /// status 41 (Unauthorized), or at once for status 87 (HistoryUnavailable) when the history of its position is + /// gone. A seek to <see cref="PollingStrategy.Next" /> reads it again after its stored offset. + /// </summary> + public bool IsResumable { get; internal init; } + + /// <summary> /// For a message that does not decode: its offset, known when its frame decoded and a later step failed, such /// as its decryption. Null when the framing of the reply is malformed, so no message can be told apart. /// </summary>
diff --git a/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj b/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj index 28dad14..5936021 100644 --- a/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj +++ b/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj
@@ -26,7 +26,7 @@ <TargetFrameworks>net8.0;net10.0</TargetFrameworks> <AssemblyName>Apache.Iggy</AssemblyName> <RootNamespace>Apache.Iggy</RootNamespace> - <Version>0.9.1-edge.8</Version> + <Version>0.9.1-edge.9</Version> <GenerateDocumentationFile>true</GenerateDocumentationFile> </PropertyGroup>
diff --git a/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/ConsumerOnReaderTests.cs b/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/ConsumerOnReaderTests.cs index eb1b018..935be0c 100644 --- a/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/ConsumerOnReaderTests.cs +++ b/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/ConsumerOnReaderTests.cs
@@ -19,8 +19,10 @@ using Apache.Iggy.Consumers; using Apache.Iggy.Contracts; using Apache.Iggy.Enums; +using Apache.Iggy.Exceptions; using Apache.Iggy.IggyClient.Implementations; using Apache.Iggy.Kinds; +using Apache.Iggy.Vsr; using Microsoft.Extensions.Logging.Abstractions; namespace Apache.Iggy.Tests.ConsumerTests; @@ -34,6 +36,10 @@ private static readonly Identifier Stream = Identifier.Numeric(1); private static readonly Identifier Topic = Identifier.Numeric(2); private static readonly TimeSpan PromptLimit = TimeSpan.FromSeconds(5); + private static readonly TimeSpan RetryInterval = TimeSpan.FromMilliseconds(20); + + /// <summary>A lost reply drops the connection, so what follows it waits for the client to connect again.</summary> + private static readonly TimeSpan LostReplyLimit = TimeSpan.FromSeconds(30); private static CancellationToken Token => TestContext.Current.CancellationToken; @@ -230,6 +236,109 @@ Assert.Equal((MessagePolling.Offset, 3ul), (error.Skip!.Value.Kind, error.Skip.Value.Value)); } + /// <summary> + /// Six refusals in a row stop the partition in the reader, and the stream ends with that error, as the Rust + /// consumer does. A seek to Next reads the partition again after its stored offset once the refusal clears. + /// </summary> + [Fact] + public async Task given_lasting_refusals_that_stop_a_partition_when_receiving_should_end_with_the_stop_until_a_seek() + { + using var cluster = new ReaderCluster(1); + cluster.Unauthorized.Add(1); + using var client = await cluster.ConnectAsync(); + await using var consumer = await StartAsync(client, CommitPolicy.Manual, 1); + + var error = await Assert.ThrowsAsync<ReadException>(() => ReceiveAsync(consumer, 1)); + cluster.Unauthorized.Clear(); + await consumer.SeekAsync(1, PollingStrategy.Next(), Token); + var received = await ReceiveAsync(consumer, 2); + + Assert.Equal((1u, VsrError.UNAUTHORIZED, true, true), (error.PartitionId, + Assert.IsType<IggyInvalidStatusCodeException>(error.InnerException).StatusCode, error.StopsPartition, + error.IsResumable)); + Assert.Equal([0ul, 1ul], received.Select(message => message.CurrentOffset)); + } + + /// <summary> + /// The history of the position is gone, so the reader stops the partition and the stream ends with that + /// error. A seek to Next reads on after the last committed message. + /// </summary> + [Fact] + public async Task given_a_lost_history_that_stops_a_partition_when_receiving_should_end_with_the_stop_until_a_seek() + { + using var cluster = new ReaderCluster(1) { Limit = 4 }; + using var client = await cluster.ConnectAsync(); + await using var consumer = await StartAsync(client, CommitPolicy.Manual, 1); + + var error = await Assert.ThrowsAsync<ReadException>(() => ReceiveUntilLostHistoryAsync(consumer, cluster)); + cluster.HistoryRefused.Clear(); + await consumer.SeekAsync(1, PollingStrategy.Next(), Token); + var resumed = await ReceiveAsync(consumer, 1); + + Assert.Equal((1u, VsrError.HISTORY_UNAVAILABLE, true, true), (error.PartitionId, + Assert.IsType<IggyInvalidStatusCodeException>(error.InnerException).StatusCode, error.StopsPartition, + error.IsResumable)); + Assert.Equal([1ul], resumed.Select(message => message.CurrentOffset)); + } + + /// <summary> + /// A committing read that lost its reply stays pending until the next read shows what it stored. A stop before + /// that read reports the lost read instead of its own error, since a seek to Next would skip what it stored. + /// </summary> + [Fact] + public async Task given_a_pending_lost_read_when_a_lost_history_stops_the_partition_should_report_the_lost_read() + { + using var cluster = new ReaderCluster(1) { Limit = 4 }; + cluster.OnLostCommittingRead = () => cluster.HistoryRefused.Add(1); + using var client = await cluster.ConnectAsync(); + await using var consumer = await StartAsync(client, CommitPolicy.OnRead, 1); + var first = await ReceiveAsync(consumer, 2); + cluster.LoseNextCommittingRead = true; + + var error = await Assert.ThrowsAsync<ReadException>(() => ReceiveAsync(consumer, 1, LostReplyLimit)); + + Assert.Equal([0ul, 1ul], first.Select(message => message.CurrentOffset)); + Assert.Equal((1u, VsrError.FETCH_PROGRESS_UNCERTAIN, true, true), (error.PartitionId, + Assert.IsType<IggyInvalidStatusCodeException>(error.InnerException).StatusCode, error.StopsPartition, + error.IsResumable)); + } + + /// <summary> + /// The first message of a last start is above the byte limit of a read, so the reader stops the partition + /// until a seek that only the application can choose. The stream ends with the error, as it does for a message + /// that does not decode, instead of waiting on a partition that never reads again. + /// </summary> + [Fact] + public async Task given_a_last_start_with_an_oversized_first_message_when_receiving_should_end_the_stream_with_its_stop() + { + using var cluster = new ReaderCluster(1) { Limit = 4, OversizedLast = true }; + using var client = await cluster.ConnectAsync(); + await using var consumer = new IggyConsumer(client, new IggyConsumerConfig + { + StreamId = Stream, + TopicId = Topic, + Consumer = Consumer.New(4), + Partitions = [1], + PollingStrategy = PollingStrategy.Last(), + PollOptions = new PollOptions { MaxWait = TimeSpan.FromMilliseconds(100) }, + RetryInterval = RetryInterval + }, NullLoggerFactory.Instance); + await consumer.InitAsync(Token); + using var timeout = CancellationTokenSource.CreateLinkedTokenSource(Token); + timeout.CancelAfter(PromptLimit); + + var error = await Assert.ThrowsAsync<ReadException>(async () => + { + await foreach (var _ in consumer.ReceiveAsync(timeout.Token)) + { + } + }); + + Assert.Equal((1u, VsrError.INVALID_SIZE_BYTES, true, false), (error.PartitionId, + Assert.IsType<IggyInvalidStatusCodeException>(error.InnerException).StatusCode, error.StopsPartition, + error.IsResumable)); + } + [Fact] public async Task given_a_cancel_after_a_message_when_receiving_should_stop_without_an_error() { @@ -344,16 +453,18 @@ TopicId = Topic, Consumer = Consumer.New(4), Partitions = partitions, - PollOptions = new PollOptions { MaxWait = TimeSpan.FromMilliseconds(100), Commit = commit } + PollOptions = new PollOptions { MaxWait = TimeSpan.FromMilliseconds(100), Commit = commit }, + RetryInterval = RetryInterval }, NullLoggerFactory.Instance); await consumer.InitAsync(Token); return consumer; } - private static async Task<List<ReceivedMessage>> ReceiveAsync(IggyConsumer consumer, int count) + private static async Task<List<ReceivedMessage>> ReceiveAsync(IggyConsumer consumer, int count, + TimeSpan? limit = null) { using var timeout = CancellationTokenSource.CreateLinkedTokenSource(Token); - timeout.CancelAfter(PromptLimit); + timeout.CancelAfter(limit ?? PromptLimit); List<ReceivedMessage> received = []; await foreach (var message in consumer.ReceiveAsync(timeout.Token)) { @@ -367,6 +478,24 @@ return received; } + /// <summary> + /// Commits offset 0 of partition 1, then loses the history of its position and reads on until the stop ends + /// the stream. + /// </summary> + private static async Task ReceiveUntilLostHistoryAsync(IggyConsumer consumer, ReaderCluster cluster) + { + using var timeout = CancellationTokenSource.CreateLinkedTokenSource(Token); + timeout.CancelAfter(PromptLimit); + await foreach (var message in consumer.ReceiveAsync(timeout.Token)) + { + if (message.CurrentOffset == 0) + { + await consumer.CommitAsync(message.Checkpoint!, Token); + cluster.HistoryRefused.Add(1); + } + } + } + private static async Task UntilAsync(Func<bool> predicate) { using var timeout = CancellationTokenSource.CreateLinkedTokenSource(Token);
diff --git a/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/ReaderCluster.cs b/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/ReaderCluster.cs index 9a72c0f..aeecd75 100644 --- a/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/ReaderCluster.cs +++ b/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/ReaderCluster.cs
@@ -124,6 +124,9 @@ internal ConcurrentBag<uint> HistoryRefused { get; } = []; + /// <summary>Partitions whose fetch section is refused as unauthorized.</summary> + internal ConcurrentBag<uint> Unauthorized { get; } = []; + internal ConcurrentDictionary<uint, ulong> Stored { get; } = new(); internal ConcurrentQueue<ReceivedFetch> Fetches { get; } = new(); @@ -290,6 +293,13 @@ continue; } + if (Unauthorized.Contains(partition)) + { + sections.Add(EmptySection(partition, 0, default, VsrError.UNAUTHORIZED)); + served = true; + continue; + } + var start = target.Kind == MessagePolling.Next ? NextOffset(partition) : target.Value; if (start >= Limit || CountLimited && held >= fetch.Count) {
diff --git a/foreign/csharp/README.md b/foreign/csharp/README.md index 207bfdd..78f16c2 100644 --- a/foreign/csharp/README.md +++ b/foreign/csharp/README.md
@@ -798,8 +798,10 @@ - `NextAsync` returns one batch of one partition, never empty. A canceled call leaves the batches in the reader. `ReadException` names the partition of an error, and the reader retries by itself. Status 87 stops a partition until `SeekAsync` when its position came from an earlier read, and so do status 80 and a refusal - that six retries did not change. Status 5009 (`ConsumerGroupPartitionNotOwned`) also drops the position of its - partition, which reads again from the start of the reader with the owner generation of a new route. + that six retries did not change. Such an error has `StopsPartition` set. `IsResumable` marks the ones that a + seek to `PollingStrategy.Next()` resumes after the stored offset: status 87 and the six refusals. Status 5009 + (`ConsumerGroupPartitionNotOwned`) also drops the position of its partition, which reads again from the start of + the reader with the owner generation of a new route. - For a message that does not decode, `ReadException` carries `Skip`, `Offset` and `Record`. `Record` is the record as the reply carried it, its 48-byte frame header, payload and user headers before decryption, so a message that does not decrypt can be kept or inspected. When the framing of the reply is malformed, `Offset` @@ -1178,7 +1180,13 @@ ends `ConsumeAsync` with its error, and its message does not count, so it is delivered again. With `CommitAfter.Marked` only `Mark(checkpoint)` counts. -A read error does not end the loop: the reader retries by itself and the error reaches `SubscribeToErrorEvents`. +A read error that the reader retries by itself does not end the loop, it reaches `SubscribeToErrorEvents`. An error +that stops a partition (`StopsPartition`) ends the loop with its `ReadException`, as a message that does not decode +does, whose `Skip` passes it: six refusals in a row such as status 41 (Unauthorized), status 87 (HistoryUnavailable) +for a position whose history is gone, and status 80 for a `Last` start whose first message is above the byte limit of +a read. The partition reads nothing until `SeekAsync`, and `IsResumable` says that a seek to `PollingStrategy.Next()` +reads it again after its stored offset. A stop while a committing read that lost its reply is still pending reports +that read as status 14010 instead, since a seek to Next would skip what it stored. `DisposeAsync` calls `ShutdownAsync`. It waits for a handler of `ConsumeAsync` that runs, unless that handler calls it, counts the last message of `ReceiveAsync` under `CommitAfter.Processing`, stores or rewinds the offsets, and leaves the group.
diff --git a/foreign/go/client/reader/reader.go b/foreign/go/client/reader/reader.go index de36ce5..af2a4ed 100644 --- a/foreign/go/client/reader/reader.go +++ b/foreign/go/client/reader/reader.go
@@ -785,9 +785,10 @@ func (r *PartitionReader) prepareRead(targets []iggcon.FetchMessagesTarget, generations map[uint32]uint64, options iggcon.DeferredPollOptions) ([]iggcon.FetchMessagesTarget, error) { type baselineResult struct { - target iggcon.FetchMessagesTarget - batch *iggcon.PolledMessage - err error + target iggcon.FetchMessagesTarget + batch *iggcon.PolledMessage + refusal error + err error } results := make(chan baselineResult, len(targets)) probes := 0 @@ -803,18 +804,19 @@ probe.MinCount = 1 sections, err := r.transport.FetchMessages(r.run, r.stream, r.topic, r.consumer, []iggcon.FetchMessagesTarget{target}, 1, false, probe) var polled *iggcon.PolledMessage + var refusal error for _, section := range sections { if section.PartitionId != target.PartitionId { continue } if section.Err != nil { - err = section.Err + refusal = section.Err break } polled, err = binaryserialization.DeserializeSectionMessages(section) break } - results <- baselineResult{target: target, batch: polled, err: err} + results <- baselineResult{target: target, batch: polled, refusal: refusal, err: err} }() } r.mu.Unlock() @@ -827,12 +829,16 @@ failure = result.err } if p != nil && p.generation == generations[result.target.PartitionId] && result.err == nil { - if result.batch != nil && len(result.batch.Messages) > 0 { + switch { + case result.refusal != nil: + // A refusal of one probe affects only its partition, so the other partitions of the read go on. + r.refuseSectionLocked(p, result.refusal, time.Now()) + case result.batch != nil && len(result.batch.Messages) > 0: first := result.batch.Messages[0].Header.Offset p.baseline = &first p.context = result.batch.Context p.position = iggcon.NextPollingStrategy().WithContext(p.context) - } else { + default: p.retryAt = time.Now().Add(r.config.retryInterval) } }
diff --git a/foreign/go/client/reader/reader_test.go b/foreign/go/client/reader/reader_test.go index a4b8eea..bd19092 100644 --- a/foreign/go/client/reader/reader_test.go +++ b/foreign/go/client/reader/reader_test.go
@@ -84,6 +84,9 @@ // caller stops waiting, as the client does. holdsCommits bool leaveErr error + // refusedProbes holds the refusals of the next probes of each partition, one per probe. + refusedProbes map[uint32][]error + probes []uint32 } func newFakeClient() *fakeClient { @@ -103,7 +106,15 @@ return nil, err } if !autoCommit && count == 1 && len(targets) == 1 && targets[0].Strategy.Kind == iggcon.POLLING_NEXT { - return []iggcon.FetchMessagesSection{section(targets[0].PartitionId, f.starts[targets[0].PartitionId])}, nil + partition := targets[0].PartitionId + f.mu.Lock() + defer f.mu.Unlock() + f.probes = append(f.probes, partition) + if refusals := f.refusedProbes[partition]; len(refusals) > 0 { + f.refusedProbes[partition] = refusals[1:] + return []iggcon.FetchMessagesSection{{PartitionId: partition, Err: refusals[0]}}, nil + } + return []iggcon.FetchMessagesSection{section(partition, f.starts[partition])}, nil } read := &fakeRead{targets: targets, count: count, commit: autoCommit, reply: make(chan fakeReply, 1)} f.reads <- read @@ -219,6 +230,12 @@ return append([]uint64(nil), f.abandoned...) } +func (f *fakeClient) recordedProbes() []uint32 { + f.mu.Lock() + defer f.mu.Unlock() + return append([]uint32(nil), f.probes...) +} + func (f *fakeClient) recordedGroupings() []time.Time { f.mu.Lock() defer f.mu.Unlock() @@ -929,6 +946,68 @@ assert.Equal(t, []uint64{0}, offsets(next(t, reader)), "the partition is read again after the report") } +// Under OnRead the partitions of a group share a read, so a refusal of the +// probe of one partition counts against that partition only. +func TestPartitionReader_StopsOnlyThePartitionWhoseProbeStaysRefusedAndKeepsReadingItsGroupUnderOnRead(t *testing.T) { + client := newFakeClient() + client.refusedProbes = map[uint32][]error{2: slices.Repeat([]error{ierror.ErrUnauthorized}, lastingRefusalsBeforeStop)} + reader, err := NewPartitionReader(context.Background(), client, identifier(t, 1), identifier(t, 2), + iggcon.NewSingleConsumer(identifier(t, 3)), WithPartitions(1, 2), + WithPollOptions(withCommit(waitOneSecond, iggcon.OnReadCommit())), WithRetryInterval(time.Millisecond)) + require.NoError(t, err) + t.Cleanup(func() { _ = reader.Close(context.Background()) }) + + var refusal *ReadError + var delivered []uint64 + reportedAfter := 0 + for refusal == nil || len(delivered) == reportedAfter { + polled := nextInBackground(reader) + var result outcome + select { + case read := <-client.reads: + require.Equal(t, []uint32{1}, partitionsOf(read), "the refused partition holds back no read of its group") + read.answer(section(1, uint64(len(delivered)))) + result = <-polled + case result = <-polled: + } + if result.err == nil { + delivered = append(delivered, offsets(result.polled)...) + continue + } + require.Nil(t, refusal, "only the refused partition is reported") + require.ErrorAs(t, result.err, &refusal) + require.Equal(t, uint32(2), *refusal.PartitionID, "only the refused partition stops") + reportedAfter = len(delivered) + } + + require.ErrorIs(t, refusal, ierror.ErrUnauthorized) + probes := client.recordedProbes() + slices.Sort(probes) + assert.Equal(t, append([]uint32{1}, slices.Repeat([]uint32{2}, lastingRefusalsBeforeStop)...), probes) +} + +// A probe refused as not the owner resets its own partition only, so the +// other partitions of its group read on from their positions. +func TestPartitionReader_KeepsThePositionsOfTheGroupOfAPartitionWhoseProbeIsRefusedAsNotOwned(t *testing.T) { + client := newFakeClient() + client.refusedProbes = map[uint32][]error{2: {ierror.ErrConsumerGroupPartitionNotOwned}} + reader := member(t, client, withCommit(waitOneSecond, iggcon.OnReadCommit()), 1) + polled := nextInBackground(reader) + client.nextRead(t).answer(section(1, 0)) + require.NoError(t, (<-polled).err) + assign(t, client, reader, 1, 2) + + polled = nextInBackground(reader) + read := client.nextRead(t) + + assert.Equal(t, []uint32{1}, partitionsOf(read), "the refused partition holds back no read of its group") + assert.Equal(t, []uint32{1, 2}, client.recordedProbes(), "the other partition reads on without a new probe") + read.answer(section(1, 1)) + result := <-polled + require.NoError(t, result.err) + assert.Equal(t, []uint64{1}, offsets(result.polled)) +} + // A section that does not decode is read again one message at a time, so // only the message that does not decode is skipped. func TestPartitionReader_ReadsAnUndecodableSectionOneMessageAtATimeAndSkipsOnlyTheBadMessage(t *testing.T) {
diff --git a/foreign/go/contracts/version.go b/foreign/go/contracts/version.go index 0f5e3d1..3ce8964 100644 --- a/foreign/go/contracts/version.go +++ b/foreign/go/contracts/version.go
@@ -17,4 +17,4 @@ package iggcon -const Version = "0.9.1-edge.8" +const Version = "0.9.1-edge.9"
diff --git a/foreign/node/README.md b/foreign/node/README.md index 283abea..d4ac866 100644 --- a/foreign/node/README.md +++ b/foreign/node/README.md
@@ -269,7 +269,9 @@ names the partition of an error, or none for an error of the whole reader, and the reader retries by itself. Error 87 stops a partition until `seek` when its position came from an earlier read, and so do error 80 and a refusal - that six retries did not change. + that six retries did not change. The error of each stop sets + `stopsPartition`, and that of an 87 or a refusal stop also sets `resumable`: + a `seek` to Next reads the partition again after its stored offset. - For a message that does not decode, `ReadError` carries `skip`, `offset` and `record`, the record as the reply carried it: its 48-byte frame header, payload and user headers, so the message can be kept or inspected. When the @@ -330,8 +332,12 @@ the first and ends as it does. `seek` drops the rest of the batch of its partition. - A read error that the reader retries by itself goes to `onError` and the - consumer goes on. A message that does not decode ends the iteration with its - `ReadError`, whose `skip` passes it with `seek`. + consumer goes on. A stop of a partition (`stopsPartition`) ends the + iteration with its `ReadError`, and so does a message that does not decode, + whose `skip` passes it with `seek`. A `resumable` stop reads again after a + `seek` to Next. A stop while a committing read that lost its reply is still + pending reports that read as 14010 instead, since a seek to Next would skip + what it stored. - `toReadable()` takes the next message only when its own reader asks for it, and destroying it shuts the consumer down.
diff --git a/foreign/node/package-lock.json b/foreign/node/package-lock.json index 81618ac..03a0e48 100644 --- a/foreign/node/package-lock.json +++ b/foreign/node/package-lock.json
@@ -1,12 +1,12 @@ { "name": "apache-iggy", - "version": "0.10.1-edge.8", + "version": "0.10.1-edge.9", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "apache-iggy", - "version": "0.10.1-edge.8", + "version": "0.10.1-edge.9", "license": "Apache-2.0", "dependencies": { "@node-rs/xxhash": "1.7.8",
diff --git a/foreign/node/package.json b/foreign/node/package.json index 6cd4029..c42d7eb 100644 --- a/foreign/node/package.json +++ b/foreign/node/package.json
@@ -1,7 +1,7 @@ { "name": "apache-iggy", "type": "module", - "version": "0.10.1-edge.8", + "version": "0.10.1-edge.9", "description": "Official Apache Iggy NodeJS SDK", "keywords": [ "iggy",
diff --git a/foreign/node/src/stream/iggy-consumer.ts b/foreign/node/src/stream/iggy-consumer.ts index 02cec9b..20a8382 100644 --- a/foreign/node/src/stream/iggy-consumer.ts +++ b/foreign/node/src/stream/iggy-consumer.ts
@@ -61,8 +61,10 @@ * `consume()` returns with success (`after: processing`), or on `mark()`, * and the latest counted position is stored at the interval or event of the * policy and on shutdown. A read error that the reader retries by itself goes - * to `onError` and the consumer goes on. A message that does not decode ends - * the iteration with its `ReadError`, whose `skip` passes it with `seek`. + * to `onError` and the consumer goes on. A stop of a partition + * (`stopsPartition`) ends the iteration with its `ReadError`, and so does a + * message that does not decode, whose `skip` passes it with `seek`. A + * `resumable` stop reads again after a `seek` to Next. */ export class IggyConsumer implements AsyncIterable<ConsumerMessage> { private uncounted?: { checkpoint: Checkpoint, endsBatch: boolean }; @@ -184,7 +186,7 @@ return await this.reader.next(); } catch (error) { if (this.stopped && !(error instanceof ReadError)) return undefined; - if (!(error instanceof ReadError) || error.skip) throw error; + if (!(error instanceof ReadError) || error.skip || error.stopsPartition) throw error; this.onError?.(error); } }
diff --git a/foreign/node/src/stream/partition-reader.test.ts b/foreign/node/src/stream/partition-reader.test.ts index 555fc75..17a9aa5 100644 --- a/foreign/node/src/stream/partition-reader.test.ts +++ b/foreign/node/src/stream/partition-reader.test.ts
@@ -41,7 +41,10 @@ const SERVED = { incarnation: 11n, ownerGeneration: 4n, metadataOp: 42n }; const BATCH_BASE_OFFSET_AT = 8; const HISTORY_UNAVAILABLE = 87; +const FETCH_PROGRESS_UNCERTAIN = 14010; const INVALID_COMMAND = 3; +const UNAUTHORIZED = 41; +const INVALID_SIZE_BYTES = 80; const CONSUMER_GROUP_PARTITION_NOT_OWNED = 5009; /** A poll body ends with the strategy kind and value, the count and the auto-commit flag. */ const POLL_SUFFIX_SIZE = 1 + 8 + 4 + 1; @@ -95,6 +98,8 @@ ownerGeneration = SERVED.ownerGeneration; /** Partitions that the member does not own, whatever a fetch target names. */ readonly notOwned = new Set<number>(); + /** The status that refuses each fetch section of a partition. */ + readonly refused = new Map<number, number>(); /** A read by offset finds nothing, as after its messages were removed. */ emptyOffsetReads = false; /** Every offset store fails with this error. */ @@ -212,6 +217,11 @@ const read = { code, partition, kind: payload[at + FETCH_TARGET_FIELD.kind], value: payload.readBigUInt64LE(at + FETCH_TARGET_FIELD.value), autocommit, count, at: Date.now() }; this.reads.push(read); + const refusal = this.refused.get(partition); + if (refusal !== undefined) { + sections.push({ partition, status: refusal, start: 0n, records: Buffer.alloc(0) }); + continue; + } if (this.history.has(partition)) { sections.push({ partition, status: HISTORY_UNAVAILABLE, start: 0n, records: Buffer.alloc(0) }); continue; @@ -1085,6 +1095,79 @@ assert.equal(first.value?.message.headers.offset, 0n); }); + for (const [stop, status] of [['six refusals in a row', UNAUTHORIZED], ['a lost history', HISTORY_UNAVAILABLE]] as const) { + it(`ends the iteration with the stop after ${stop}, and a seek to Next reads the partition again`, async () => { + const topic = new FakeTopic(4n); + topic.stored.set(1, 1n); + const client = topic.client(); + const errors: ReadError[] = []; + const consumer = new IggyConsumer(await PartitionReader.open(async () => client, { + streamId: 1, topicId: 2, consumer: { kind: ConsumerKind.Single, id: 4 }, partitions: [1], + options: { maxWait: 100 }, retryInterval: 5 + }), undefined, (error) => { errors.push(error); }); + try { + const messages = consumer[Symbol.asyncIterator](); + const before = [await messages.next(), await messages.next()]; + topic.refused.set(1, status); + const stopped = await messages.next().then(() => undefined, (error: unknown) => error); + topic.refused.delete(1); + await consumer.seek(1, PollingStrategy.Next); + const resumed = await consumer[Symbol.asyncIterator]().next(); + + assert.deepEqual(before.map((read) => read.value?.message.headers.offset), [2n, 3n]); + assert.ok(stopped instanceof ReadError, `the iteration did not end with the stop: ${String(stopped)}`); + assert.deepEqual([stopped.partitionId, (stopped.error as { errorCode: number }).errorCode, + stopped.stopsPartition, stopped.resumable], [1, status, true, true]); + assert.equal(resumed.value?.message.headers.offset, 2n, 'a seek to Next reads after the stored offset'); + assert.deepEqual(errors, [], 'a stop is thrown, not handed to onError'); + } finally { + await consumer.shutdown(); + } + }); + } + + it('reports a pending lost read instead of the stop that ends the iteration', async () => { + const topic = new FakeTopic(4n); + topic.onLostCommittingRead = () => topic.history.add(1); + const consumer = new IggyConsumer(await open(topic, [1], { commit: CommitPolicy.OnRead })); + const messages = consumer[Symbol.asyncIterator](); + try { + const first = [await messages.next(), await messages.next()]; + topic.loseNextCommittingRead = true; + const stopped = await messages.next().then(() => undefined, (error: unknown) => error); + + assert.deepEqual(first.map((read) => read.value?.message.headers.offset), [0n, 1n]); + assert.ok(stopped instanceof ReadError, `the iteration did not end with the stop: ${String(stopped)}`); + assert.deepEqual([stopped.partitionId, (stopped.error as { errorCode: number }).errorCode, + stopped.stopsPartition, stopped.resumable], [1, FETCH_PROGRESS_UNCERTAIN, true, true], + 'a seek to Next would skip what the lost read stored, so the stop reports that read'); + } finally { + await consumer.shutdown(); + } + }); + + for (const read of ['iterator', 'consume', 'readable'] as const) { + it(`ends the ${read} with the error of a stop that the consumer does not resume`, async () => { + const topic = new FakeTopic(4n); + topic.refused.set(1, INVALID_SIZE_BYTES); + const errors: ReadError[] = []; + const consumer = new IggyConsumer(await open(topic, [1]), undefined, (error) => { errors.push(error); }); + try { + const ended = read === 'iterator' ? consumer[Symbol.asyncIterator]().next() + : read === 'consume' ? consumer.consume(() => undefined) + : once(consumer.toReadable().resume(), 'error').then(([error]) => Promise.reject(error)); + const outcome = await Promise.race([ended.then(() => 'ended', (error: unknown) => error), + new Promise((resolve) => setTimeout(resolve, 2_000, 'stalled'))]); + + assert.ok(outcome instanceof ReadError, `the ${read} did not end with a read error: ${String(outcome)}`); + assert.deepEqual([outcome.partitionId, (outcome.error as { errorCode: number }).errorCode, + outcome.stopsPartition, outcome.resumable, errors.length], [1, INVALID_SIZE_BYTES, true, false, 0]); + } finally { + await consumer.shutdown(); + } + }); + } + it('shuts down when its readable is destroyed while a read waits for messages', async () => { const topic = new FakeTopic(0n); const consumer = new IggyConsumer(await open(topic, [1]));
diff --git a/foreign/node/src/stream/partition-reader.ts b/foreign/node/src/stream/partition-reader.ts index 1b12ce8..1558eda 100644 --- a/foreign/node/src/stream/partition-reader.ts +++ b/foreign/node/src/stream/partition-reader.ts
@@ -101,7 +101,14 @@ * reader reads the message again until a `seek` to this continuation * skips it. The messages before it arrive in a batch before this error. */ - readonly skip?: Continuation + readonly skip?: Continuation, + /** The reader stopped the partition, which it reads again only after a `seek`. */ + readonly stopsPartition = false, + /** + * The reader stopped the partition after refusals in a row or a lost + * history. A `seek` to Next reads it again after its stored offset. + */ + readonly resumable = false ) { super(partitionId === undefined ? `reading failed: ${String(error)}` : `reading partition ${partitionId} failed: ${String(error)}`, { cause: error }); @@ -1027,8 +1034,16 @@ state.backlogged = undefined; const now = Date.now(); const code = errorCode(error); - const report = (stop: boolean) => { - const readError = new ReadError(target.partition, error); + const report = (stop: boolean, resumable = false) => { + if (stop) this.giveUpGap(target.partition, state); + // A committing read that lost its reply stays pending until the next read shows what it stored, and a seek + // to Next after the stop would skip that range without a word, so the stop reports the lost read instead. + const reported = stop && state.lost + ? responseError(COMMAND_CODE.PollMessagesDeferred, FETCH_PROGRESS_UNCERTAIN, + `a lost read may have stored offsets of partition ${target.partition} from ${state.position.value}, ` + + `and the partition stopped on: ${String(error)}`) + : error; + const readError = new ReadError(target.partition, reported, undefined, stop, resumable); if (stop) state.stopped = readError; this.enqueue({ partitionId: target.partition, generation: target.generation, error: readError, messages: 0, bytes: 0 }); @@ -1049,13 +1064,13 @@ return; } if (code === HISTORY_UNAVAILABLE || code === INVALID_SIZE_BYTES) { - report(true); + report(true, code === HISTORY_UNAVAILABLE); return; } if (code !== undefined && LASTING_REFUSALS.has(code)) { state.lasting += 1; if (state.lasting >= MAX_LASTING_REFUSALS) { - report(true); + report(true, true); return; } state.retryAt = now + this.retryInterval * 2 ** (state.lasting - 1);
diff --git a/foreign/php/Cargo.toml b/foreign/php/Cargo.toml index cd25bff..b1523e3 100644 --- a/foreign/php/Cargo.toml +++ b/foreign/php/Cargo.toml
@@ -17,7 +17,7 @@ [package] name = "iggy-php" -version = "0.1.10" +version = "0.1.11" edition = "2024" authors = ["Iggy Committers <dev@iggy.apache.org>"] license = "Apache-2.0"
diff --git a/foreign/php/iggy-php.stubs.php b/foreign/php/iggy-php.stubs.php index f6a12ce..0b7a368 100644 --- a/foreign/php/iggy-php.stubs.php +++ b/foreign/php/iggy-php.stubs.php
@@ -688,11 +688,12 @@ * then stays uncounted under CommitAfter::Processing, so the next run reads it again. * * The consumer retries a failed read by itself and does not throw it. An error that ends - * the run throws Iggy\Exception\ReadException: HistoryUnavailable (87), which stops its - * partition until seek(), and under CommitPolicy::onRead() an unknown outcome, which - * stops its partition until its operation is abandoned and the partition is sought. A - * blocked lane, a shut down client and any error while the client cannot reconnect by - * itself also end the run. + * the run throws Iggy\Exception\ReadException: HistoryUnavailable (87) and six refusals + * in a row that a retry does not change, such as Unauthorized (41) or a missing partition, + * each of which stops its partition until seek(), and under CommitPolicy::onRead() an + * unknown outcome, which stops its partition until its operation is abandoned and the + * partition is sought. A blocked lane, a shut down client and any error while the client + * cannot reconnect by itself also end the run. * * With `idle_timeout_micros`, a run that gets no message for that long ends and returns * the count of the messages it consumed, so a single-threaded worker regains control on an @@ -748,7 +749,8 @@ * Under CommitPolicy::auto(), a message counts when the iterator yields it with * CommitAfter::Delivery, and when the iterator moves past it with CommitAfter::Processing. A * failed read throws Iggy\Exception\ReadException. The consumer stays usable and retries by - * itself. HistoryUnavailable (87) stops its partition until seek(). + * itself. HistoryUnavailable (87) and six refusals in a row that a retry does not change, + * such as Unauthorized (41), stop their partition until seek(). * * @param mixed $idle_timeout_micros * @return \Iggy\MessageIterator @@ -2755,6 +2757,14 @@ */ class ReadException extends \Iggy\Exception\IggyException { /** + * Whether the reader stopped the partition after six refusals in a row that a retry does + * not change, such as Unauthorized (41). A seek of the partition reads it again. + * + * @var bool + */ + public readonly bool $is_lasting_refusal; + + /** * The partition of the error, or null for an error of the whole reader. * * @var int|null
diff --git a/foreign/php/src/consumer.rs b/foreign/php/src/consumer.rs index 0099095..93caba0 100644 --- a/foreign/php/src/consumer.rs +++ b/foreign/php/src/consumer.rs
@@ -172,11 +172,12 @@ /// then stays uncounted under CommitAfter::Processing, so the next run reads it again. /// /// The consumer retries a failed read by itself and does not throw it. An error that ends - /// the run throws Iggy\Exception\ReadException: HistoryUnavailable (87), which stops its - /// partition until seek(), and under CommitPolicy::onRead() an unknown outcome, which - /// stops its partition until its operation is abandoned and the partition is sought. A - /// blocked lane, a shut down client and any error while the client cannot reconnect by - /// itself also end the run. + /// the run throws Iggy\Exception\ReadException: HistoryUnavailable (87) and six refusals + /// in a row that a retry does not change, such as Unauthorized (41) or a missing partition, + /// each of which stops its partition until seek(), and under CommitPolicy::onRead() an + /// unknown outcome, which stops its partition until its operation is abandoned and the + /// partition is sought. A blocked lane, a shut down client and any error while the client + /// cannot reconnect by itself also end the run. /// /// With `idle_timeout_micros`, a run that gets no message for that long ends and returns /// the count of the messages it consumed, so a single-threaded worker regains control on an @@ -253,7 +254,8 @@ /// Under CommitPolicy::auto(), a message counts when the iterator yields it with /// CommitAfter::Delivery, and when the iterator moves past it with CommitAfter::Processing. A /// failed read throws Iggy\Exception\ReadException. The consumer stays usable and retries by - /// itself. HistoryUnavailable (87) stops its partition until seek(). + /// itself. HistoryUnavailable (87) and six refusals in a row that a retry does not change, + /// such as Unauthorized (41), stop their partition until seek(). pub fn iter_messages(&self, idle_timeout_micros: Option<&Zval>) -> PhpResult<MessageIterator> { let idle_timeout = optional_unsigned::<u64>(idle_timeout_micros, "idle_timeout_micros")? .map(Duration::from_micros);
diff --git a/foreign/php/src/error.rs b/foreign/php/src/error.rs index 9b91c19..81ae5d5 100644 --- a/foreign/php/src/error.rs +++ b/foreign/php/src/error.rs
@@ -123,6 +123,7 @@ pub struct ReadException { partition_id: Option<u32>, skip: Option<PollingStrategy>, + lasting_refusal: bool, } #[php_impl] @@ -139,6 +140,13 @@ pub fn skip(&self) -> Option<PollingStrategy> { self.skip.clone() } + + /// Whether the reader stopped the partition after six refusals in a row that a retry does + /// not change, such as Unauthorized (41). A seek of the partition reads it again. + #[php(getter, change_case = "snake_case")] + pub fn is_lasting_refusal(&self) -> bool { + self.lasting_refusal + } } pub(crate) trait IntoPhpException { @@ -167,6 +175,7 @@ let code = self.error.as_code() as i32; let exception = ReadException { partition_id: self.partition_id, + lasting_refusal: self.is_lasting_refusal(), skip: self.skip.map(PollingStrategy::from), };
diff --git a/foreign/python/Cargo.toml b/foreign/python/Cargo.toml index d737d65..d01a77f 100644 --- a/foreign/python/Cargo.toml +++ b/foreign/python/Cargo.toml
@@ -17,7 +17,7 @@ [package] name = "apache-iggy" -version = "0.9.1-dev9" +version = "0.9.1-dev10" edition = "2024" authors = ["Iggy Committers <dev@iggy.apache.org>"] license = "Apache-2.0" @@ -37,7 +37,7 @@ [dependencies] bytes = "1.12.1" futures = "0.3.34" -iggy = { path = "../../core/sdk", version = "0.11.1-edge.8" } +iggy = { path = "../../core/sdk", version = "0.11.1-edge.9" } paste = "1" pyo3 = { version = "0.29.3", features = ["experimental-async"] } pyo3-async-runtimes = { version = "0.29.0", features = [
diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 349f7be..4cd6416 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi
@@ -2424,7 +2424,9 @@ `CommitAfter.Delivery`, and when the next iteration starts with `CommitAfter.Processing`. After another client deleted and created a partition again, one iteration raises `ReadError` with `HistoryUnavailable` (87), and the partition reads nothing until - `seek()` picks a start in its new history. + `seek()` picks a start in its new history. After six refusals in a row that a retry does + not change, such as `Unauthorized` (41), one iteration raises a `ReadError` whose + `is_lasting_refusal` is true, and the partition reads nothing until `seek()`. A cancelled iteration leaves its message for the next one. Iterating in a `consume_messages()` callback of this consumer raises `RuntimeError`. """ @@ -2446,7 +2448,8 @@ that raises is logged and the run goes on. A failed read is logged, and the consumer retries by itself. The run raises `ReadError` if the client cannot go on without the caller, or if a read stops its partition until - `seek()`: `HistoryUnavailable` (87), or an unknown outcome under + `seek()`: `HistoryUnavailable` (87), six refusals in a row that a retry does not change, + such as `Unauthorized` (41) or a missing partition, or an unknown outcome under `CommitPolicy.OnRead()`. For a stopped partition, `ReadError.partition_id` names it. To resume it, abandon the operation of an unknown outcome, seek the partition and consume again. @@ -4442,6 +4445,12 @@ @property def offset(self) -> builtins.int | None: ... @property + def is_lasting_refusal(self) -> builtins.bool: + r""" + Whether the reader stopped the partition after six refusals in a row that a retry does + not change, such as `Unauthorized` (41). A `seek()` of the partition reads it again. + """ + @property def error(self) -> IggyError: r""" The error as the client raises it for any other call: an `IggyError`, or a
diff --git a/foreign/python/pyproject.toml b/foreign/python/pyproject.toml index 3091074..b6b9a22 100644 --- a/foreign/python/pyproject.toml +++ b/foreign/python/pyproject.toml
@@ -22,7 +22,7 @@ [project] name = "apache-iggy" requires-python = ">=3.11" -version = "0.9.1.dev9" +version = "0.9.1.dev10" 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/src/client.rs b/foreign/python/src/client.rs index e91ed73..bd4ffbc 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs
@@ -124,6 +124,10 @@ #[pyo3(get)] offset: Option<u64>, record: Option<Bytes>, + /// Whether the reader stopped the partition after six refusals in a row that a retry does + /// not change, such as `Unauthorized` (41). A `seek()` of the partition reads it again. + #[pyo3(get)] + is_lasting_refusal: bool, } #[gen_stub_pymethods] @@ -189,12 +193,14 @@ let error = self.0; let message = error.to_string(); let coded = PyIggyError::from(&error.error); + let is_lasting_refusal = error.is_lasting_refusal(); let read_error = ReadError { partition_id: error.partition_id, error: runtime_error_value(py, error.error), skip: error.skip, offset: error.offset, record: error.record, + is_lasting_refusal, }; exception_value( py, @@ -2593,6 +2599,37 @@ } #[test] + fn given_read_error_of_a_lasting_refusal_when_mapping_should_flag_only_the_stop() { + Python::initialize(); + Python::attach(|py| { + for (partition_id, error, flagged) in [ + (Some(1), IggyError::Unauthorized, true), + (None, IggyError::Unauthorized, false), + (Some(1), IggyError::HistoryUnavailable, false), + ] { + let error = to_read_error(RustReadError { + partition_id, + error, + skip: None, + offset: None, + record: None, + }); + + let value = error.value(py); + assert_eq!( + value + .getattr("is_lasting_refusal") + .unwrap() + .extract::<bool>() + .unwrap(), + flagged + ); + assert!(value.setattr("is_lasting_refusal", false).is_err()); + } + }); + } + + #[test] fn given_read_error_of_the_whole_reader_when_mapping_should_have_no_partition_or_skip() { Python::initialize(); Python::attach(|py| {
diff --git a/foreign/python/src/consumer.rs b/foreign/python/src/consumer.rs index 9739af5..fbd8793 100644 --- a/foreign/python/src/consumer.rs +++ b/foreign/python/src/consumer.rs
@@ -268,7 +268,9 @@ /// `CommitAfter.Delivery`, and when the next iteration starts with `CommitAfter.Processing`. /// After another client deleted and created a partition again, one iteration raises /// `ReadError` with `HistoryUnavailable` (87), and the partition reads nothing until - /// `seek()` picks a start in its new history. + /// `seek()` picks a start in its new history. After six refusals in a row that a retry does + /// not change, such as `Unauthorized` (41), one iteration raises a `ReadError` whose + /// `is_lasting_refusal` is true, and the partition reads nothing until `seek()`. /// A cancelled iteration leaves its message for the next one. Iterating in a /// `consume_messages()` callback of this consumer raises `RuntimeError`. #[gen_stub(override_return_type(type_repr="collections.abc.AsyncIterator[ReceiveMessage]", imports=("collections.abc")))] @@ -290,7 +292,8 @@ /// that raises is logged and the run goes on. /// A failed read is logged, and the consumer retries by itself. The run raises `ReadError` /// if the client cannot go on without the caller, or if a read stops its partition until - /// `seek()`: `HistoryUnavailable` (87), or an unknown outcome under + /// `seek()`: `HistoryUnavailable` (87), six refusals in a row that a retry does not change, + /// such as `Unauthorized` (41) or a missing partition, or an unknown outcome under /// `CommitPolicy.OnRead()`. For a stopped partition, `ReadError.partition_id` names it. /// To resume it, abandon the operation of an unknown outcome, seek the partition and consume /// again.
diff --git a/foreign/python/uv.lock b/foreign/python/uv.lock index 0ab7ee1..dff5797 100644 --- a/foreign/python/uv.lock +++ b/foreign/python/uv.lock
@@ -11,7 +11,7 @@ [[package]] name = "apache-iggy" -version = "0.9.1.dev9" +version = "0.9.1.dev10" source = { editable = "." } [package.optional-dependencies]
diff --git a/foreign/swift/README.md b/foreign/swift/README.md index 5cac0dc..a9d6040 100644 --- a/foreign/swift/README.md +++ b/foreign/swift/README.md
@@ -411,7 +411,11 @@ A handler that throws leaves its message uncounted and ends `consumeMessages(_:)` with its error, so the message is read again by the next run. Under the other policies a handler that throws is logged and reading goes on. -`pollingStrategy` says where a partition without a position starts. A failed read is thrown as a `ReadError` and leaves the consumer usable: a caller of `next()` can call it again, and `consumeMessages(_:)` logs it and keeps reading until the client cannot come back on its own. Without `allowReplay` a message at or below the last one handed over from the same partition history is skipped, unless `seek(partitionID:to:)` moved the partition back. +`pollingStrategy` says where a partition without a position starts. A failed read is thrown as a `ReadError` and leaves the consumer usable: a caller of `next()` can call it again, and `consumeMessages(_:)` logs it and keeps reading. + +`consumeMessages(_:)` ends with the error once the client cannot come back on its own, and for a message that does not decode or `historyUnavailable`, which stop the partition until a seek. + +It also ends with `partitionLaneBlocked` under `.onRead`: a lost committing read of another caller blocks the partition until `abandonOperation(_:)` takes that operation, after which the reader reads it again by itself. Without `allowReplay` a message at or below the last one handed over from the same partition history is skipped, unless `seek(partitionID:to:)` moved the partition back. `shutdown()` counts the last message under `.processing` and closes the reader within `offsetDrainTimeout`: it stores the counted positions under `.auto`, moves the offset of each partition back to the last message handed over under `.onRead`, and leaves the group.
diff --git a/foreign/swift/Sources/Iggy/Consumer/IggyConsumer.swift b/foreign/swift/Sources/Iggy/Consumer/IggyConsumer.swift index 724c00f..a131d54 100644 --- a/foreign/swift/Sources/Iggy/Consumer/IggyConsumer.swift +++ b/foreign/swift/Sources/Iggy/Consumer/IggyConsumer.swift
@@ -209,13 +209,21 @@ } /// Hands every message to `handler` until the consumer shuts down, the - /// task is cancelled, or a read error ends the reading: a client that is - /// shut down or cannot come back without the caller, or an invalid - /// address. Other read errors are logged and reading goes on. Under - /// ``CommitAfter/processing`` a message counts once its handler returned, - /// and a handler that throws leaves its message uncounted and ends the - /// reading with its error. Under any other policy a handler that throws - /// is logged. + /// task is cancelled, or a read error ends the reading with that error. + /// A message that does not decode ends it, and so does + /// ``IggyErrorCode/historyUnavailable`` for a position read in a history + /// that is gone. Both stop their partition until a + /// ``seek(partitionID:to:)``, and a seek to ``ReadError/skip`` steps over + /// the message. So does ``IggyErrorCode/partitionLaneBlocked`` under + /// ``CommitPolicy/onRead``: a lost committing read of another caller + /// blocks the partition until ``IggyClient/abandonOperation(_:)`` takes + /// that operation, after which the reader reads it again by itself. A + /// client that is shut down or cannot come back without + /// the caller ends it, and so does an invalid address. Other read errors + /// are logged and reading goes on. Under ``CommitAfter/processing`` a + /// message counts once its handler returned, and a handler that throws + /// leaves its message uncounted and ends the reading with its error. + /// Under any other policy a handler that throws is logged. public func consumeMessages(_ handler: @Sendable (ReceivedMessage) async throws -> Void) async throws { await countUnprocessed() let processing = configuration.pollOptions.autoCommit?.after == .processing @@ -228,7 +236,7 @@ received = next } catch let error as ReadError { let recovers = await client.canRecover - if Self.endsConsumption(error.error) || !recovers { + if error.stopsPartition || Self.endsConsumption(error.error) || !recovers { logger.error("Reading stopped", metadata: metadata.merging(["error": "\(error)"]) { $1 }) throw error }
diff --git a/foreign/swift/Sources/Iggy/Consumer/PartitionReader.swift b/foreign/swift/Sources/Iggy/Consumer/PartitionReader.swift index 0823678..e8a003a 100644 --- a/foreign/swift/Sources/Iggy/Consumer/PartitionReader.swift +++ b/foreign/swift/Sources/Iggy/Consumer/PartitionReader.swift
@@ -87,13 +87,21 @@ public let offset: UInt64? /// The record of the message that does not decode, as the server sent it. public let raw: [UInt8]? + /// Whether the partition reads nothing until a + /// ``PartitionReader/seek(partitionID:to:)``: a message that does not + /// decode, or a position whose history is gone. A partition that a lost + /// committing read of another caller blocks (``IggyErrorCode/partitionLaneBlocked`` + /// under ``CommitPolicy/onRead``) reads again by itself once + /// ``IggyClient/abandonOperation(_:)`` took that operation. + public let stopsPartition: Bool - init(partitionID: UInt32?, error: IggyError, skip: PollingStrategy? = nil, offset: UInt64? = nil, raw: [UInt8]? = nil) { + init(partitionID: UInt32?, error: IggyError, skip: PollingStrategy? = nil, offset: UInt64? = nil, raw: [UInt8]? = nil, stopsPartition: Bool = false) { self.partitionID = partitionID self.error = error self.skip = skip self.offset = offset self.raw = raw + self.stopsPartition = stopsPartition } public var description: String { @@ -1566,7 +1574,7 @@ isolations[partition] = nil stopped.insert(partition) let skip = PollingStrategy.continuation(partitionID: partition, kind: .offset, value: isolation.first + 1, context: isolation.context) - report(ReadError(partitionID: partition, error: Self.iggyError(error), skip: skip, offset: isolation.first, raw: raw)) + report(ReadError(partitionID: partition, error: Self.iggyError(error), skip: skip, offset: isolation.first, raw: raw, stopsPartition: true)) return } if request.kind == .gap, let gap = gaps[partition] { @@ -1583,7 +1591,7 @@ return } stopped.insert(partition) - report(ReadError(partitionID: partition, error: Self.iggyError(error), raw: raw)) + report(ReadError(partitionID: partition, error: Self.iggyError(error), raw: raw, stopsPartition: true)) } /// A message that an isolation delivered moves on the gap it lies in, or @@ -1659,6 +1667,7 @@ /// ``next()``. private func failed(_ partitions: [UInt32], with error: IggyError) { let until = ContinuousClock.now + configuration.retryInterval + var stops = false switch error.code { case .notPrimary: groupingEpoch += 1 @@ -1677,7 +1686,7 @@ for partition in partitions { if positions[partition]?.context != nil { stopped.insert(partition) - report(ReadError(partitionID: partition, error: error)) + report(ReadError(partitionID: partition, error: error, stopsPartition: true)) } else { forget(partition) pausedUntil[partition] = until @@ -1701,6 +1710,7 @@ stopped.insert(partition) blockedBy[partition] = error.operationID } + stops = true case .disconnected, .unauthenticated, .staleClient: joinedGroup = false default: @@ -1710,7 +1720,7 @@ pausedUntil[partition] = until } logger.error("Failed to read messages", metadata: metadata.merging(["partitions": "\(partitions)", "error": "\(error)"]) { $1 }) - report(ReadError(partitionID: partitions.count == 1 ? partitions.first : nil, error: error)) + report(ReadError(partitionID: partitions.count == 1 ? partitions.first : nil, error: error, stopsPartition: stops)) } /// Queues an error for ``next()``. The latest one of each partition and
diff --git a/foreign/swift/Sources/Iggy/Iggy.swift b/foreign/swift/Sources/Iggy/Iggy.swift index b0de655..3addcb8 100644 --- a/foreign/swift/Sources/Iggy/Iggy.swift +++ b/foreign/swift/Sources/Iggy/Iggy.swift
@@ -21,5 +21,5 @@ public static let sdkName = "swift-sdk" /// Version of this SDK as reported to the server. Provisional until the /// release wiring lands; the release tooling reads this constant. - public static let sdkVersion = "0.1.8" + public static let sdkVersion = "0.1.9" }
diff --git a/foreign/swift/Tests/IggyTests/ConsumerTests.swift b/foreign/swift/Tests/IggyTests/ConsumerTests.swift index 8c4b3ad..63ecd29 100644 --- a/foreign/swift/Tests/IggyTests/ConsumerTests.swift +++ b/foreign/swift/Tests/IggyTests/ConsumerTests.swift
@@ -1370,6 +1370,41 @@ await server.stop() } + /// A position whose history is gone stops its partition until a seek, so + /// the handler run ends with that error instead of going on without the + /// partition. + @Test func aReadErrorThatStopsItsPartitionEndsConsumeMessages() async throws { + let server = try await FakeVSRServer.started() + let topic = FakeTopic(ends: [1: 2]) + await serve(server, topic: topic) { request in + guard request.code == CommandCode.fetchMessages.rawValue, let fetch = try? request.fetch, fetch.targets.allSatisfy({ $0.value == 1 }) + else { + return nil + } + return .ok(FakeVSRServer.fetchedBody([.init(partitionID: 1, status: IggyErrorCode.historyUnavailable.rawValue)])) + } + let client = try await signedInClient(server) + let consumer = try client.consumer( + name: "reader", streamID: 1, topicID: 2, partitions: [1], + configuration: ConsumerConfiguration(pollingStrategy: .first, pollOptions: PollOptions(count: 1, maxWait: .milliseconds(200)))) + try await consumer.initialize() + let handled = Locked<[UInt64]>([]) + + let error = await #expect(throws: ReadError.self, "the run did not end within 2 seconds") { + try await withDeadline(.now + .seconds(2)) { + try await consumer.consumeMessages { received in + handled.withLock { $0.append(received.message.offset) } + } + } + } + + #expect(error?.partitionID == 1 && error?.error.code == .historyUnavailable) + #expect(handled.withLock { $0 } == [0]) + await consumer.shutdown() + await client.shutdown() + await server.stop() + } + /// A seek can replace the generation of a partition while the consumer /// waits for the batch it then buffers. That batch is dropped, and the /// batches of the other partitions still arrive.
diff --git a/foreign/swift/Tests/IggyTests/Fixtures/golden.json b/foreign/swift/Tests/IggyTests/Fixtures/golden.json index ec2fb87..579d359 100644 --- a/foreign/swift/Tests/IggyTests/Fixtures/golden.json +++ b/foreign/swift/Tests/IggyTests/Fixtures/golden.json
@@ -935,7 +935,7 @@ "polling.next": "050000000000000000", "polling.offset.100": "016400000000000000", "polling.timestamp.1700000000000": "020068e5cf8b010000", - "request.bind_session": "023000000973776966742d73646b05302e312e3801000000000000000000000000000000640000000000000078000000000000005a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a", + "request.bind_session": "023000000973776966742d73646b05302e312e3901000000000000000000000000000000640000000000000078000000000000005a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a", "request.change_password": "010401000000086f6c642d706173730c6e65772d706173732d313233", "request.create_consumer_group": "01040100000001040200000003677270", "request.create_partitions": "01040100000001040200000005000000", @@ -976,8 +976,8 @@ "request.get_users": "", "request.join_consumer_group": "010401000000010402000000010403000000", "request.leave_consumer_group": "010401000000010402000000020167", - "request.login_register": "023000000973776966742d73646b05302e312e385a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a0469676779046967677900000000", - "request.login_register_with_pat": "023000000973776966742d73646b05302e312e385a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a107061742d61626331323364656634353600000000", + "request.login_register": "023000000973776966742d73646b05302e312e395a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a0469676779046967677900000000", + "request.login_register_with_pat": "023000000973776966742d73646b05302e312e395a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a107061742d61626331323364656634353600000000", "request.logout_user": "", "request.ping": "", "request.poll_messages": "0101040100000001040a00000001041400000001050000000164000000000000003200000001", @@ -1032,7 +1032,7 @@ "response.user_details.no_permissions": "0100000000e090663c130600010561646d696e000000000000000000000000", "stream_key.fingerprint": "8441e38538cbfa57", "user_headers.sample": "0207000000617474656d70740b0400000003000000020800000074726163652d69640203000000616263", - "version_info": "023000000973776966742d73646b05302e312e38", + "version_info": "023000000973776966742d73646b05302e312e39", "vsr.eviction.incompatible_protocol": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001000000000000000000000d000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab89674523010230000002300000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000e", "vsr.eviction.stale_client": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001000000000000000000000d000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab89674523010000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000d", "vsr.reply.committed": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000c0100000000000000000000080000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab896745230100000000000000000c0000000000000000000000000000000700000000000000800000000000000000000000010000000000000000000000000000000000000000000000000000000000000000000000",