From 0bccdce5c0d2ef7f0345eda0a73ed41e902bb9b1 Mon Sep 17 00:00:00 2001 From: Ethan Lin Date: Wed, 19 Aug 2026 23:35:44 +0800 Subject: [PATCH 1/4] feat(common): add NonZeroIggyDuration Intervals that pace a loop panic or spin when they are zero. A dedicated type lets those fields reject the zero at construction instead of at every call site, and rejects the 'none', 'disabled' and 'unlimited' aliases that IggyDuration parses as zero. --- core/common/src/lib.rs | 1 + core/common/src/utils/mod.rs | 1 + core/common/src/utils/non_zero_duration.rs | 255 +++++++++++++++++++++ 3 files changed, 257 insertions(+) create mode 100644 core/common/src/utils/non_zero_duration.rs diff --git a/core/common/src/lib.rs b/core/common/src/lib.rs index 777b5ef276..8165d6965d 100644 --- a/core/common/src/lib.rs +++ b/core/common/src/lib.rs @@ -131,6 +131,7 @@ pub use utils::expiry::IggyExpiry; pub use utils::hash::*; pub use utils::net::validate_api_url; pub use utils::net::validate_server_address; +pub use utils::non_zero_duration::{NonZeroDurationError, NonZeroIggyDuration}; pub use utils::personal_access_token_expiry::PersonalAccessTokenExpiry; pub use utils::random_id; pub use utils::serde_secret; diff --git a/core/common/src/utils/mod.rs b/core/common/src/utils/mod.rs index 4ac92d81cd..72280c70c9 100644 --- a/core/common/src/utils/mod.rs +++ b/core/common/src/utils/mod.rs @@ -22,6 +22,7 @@ pub(crate) mod duration; pub(crate) mod expiry; pub(crate) mod hash; pub(crate) mod net; +pub(crate) mod non_zero_duration; pub(crate) mod personal_access_token_expiry; pub mod random_id; pub mod serde_secret; diff --git a/core/common/src/utils/non_zero_duration.rs b/core/common/src/utils/non_zero_duration.rs new file mode 100644 index 0000000000..167efa1daa --- /dev/null +++ b/core/common/src/utils/non_zero_duration.rs @@ -0,0 +1,255 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::IggyDuration; +use serde::de::{Error as DeError, Visitor}; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use std::{ + error::Error, + fmt::{Display, Formatter}, + str::FromStr, + time::Duration, +}; + +/// A duration that is guaranteed to be greater than zero. +/// +/// Intervals that pace a loop - heartbeats, reconnection and retry delays - turn into +/// a busy loop or a `tokio::time::interval` panic when they are zero. Such fields hold +/// this type so the zero is rejected where the value is built, not where it is awaited. +/// +/// `IggyDuration::from_str` maps `0`, `none`, `disabled` and `unlimited` to the same +/// zero, so all four are rejected here. +/// +/// # Example +/// +/// ``` +/// use iggy_common::{IggyDuration, NonZeroIggyDuration, NonZeroDurationError}; +/// use std::str::FromStr; +/// +/// let interval = NonZeroIggyDuration::from_str("1s").unwrap(); +/// assert_eq!(1, interval.as_secs()); +/// assert_eq!("1s", format!("{}", interval)); +/// +/// assert_eq!(Err(NonZeroDurationError::Zero), NonZeroIggyDuration::from_str("none")); +/// assert_eq!( +/// Err(NonZeroDurationError::Zero), +/// NonZeroIggyDuration::try_from(IggyDuration::from(0_u64)), +/// ); +/// ``` +#[derive(Debug, Clone, Copy, Eq, PartialEq)] +pub struct NonZeroIggyDuration { + duration: IggyDuration, +} + +/// The reason a value could not become a `NonZeroIggyDuration`. +#[derive(Debug, Clone, PartialEq)] +pub enum NonZeroDurationError { + /// The value parsed or converted to zero. + Zero, + /// The text is not a duration `humantime` understands. + InvalidFormat(humantime::DurationError), +} + +impl NonZeroIggyDuration { + pub const ONE_SECOND: NonZeroIggyDuration = NonZeroIggyDuration { + duration: IggyDuration::ONE_SECOND, + }; + + pub fn get(&self) -> IggyDuration { + self.duration + } + + pub fn get_duration(&self) -> Duration { + self.duration.get_duration() + } + + pub fn as_human_time_string(&self) -> String { + self.duration.as_human_time_string() + } + + pub fn as_secs(&self) -> u32 { + self.duration.as_secs() + } + + pub fn as_micros(&self) -> u64 { + self.duration.as_micros() + } +} + +impl Display for NonZeroDurationError { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + match self { + NonZeroDurationError::Zero => write!(f, "duration must be greater than zero"), + NonZeroDurationError::InvalidFormat(error) => write!(f, "invalid duration: {error}"), + } + } +} + +impl Error for NonZeroDurationError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + match self { + NonZeroDurationError::Zero => None, + NonZeroDurationError::InvalidFormat(error) => Some(error), + } + } +} + +impl From for NonZeroDurationError { + fn from(error: humantime::DurationError) -> Self { + NonZeroDurationError::InvalidFormat(error) + } +} + +impl TryFrom for NonZeroIggyDuration { + type Error = NonZeroDurationError; + + fn try_from(duration: IggyDuration) -> Result { + if duration.is_zero() { + return Err(NonZeroDurationError::Zero); + } + + Ok(NonZeroIggyDuration { duration }) + } +} + +impl TryFrom for NonZeroIggyDuration { + type Error = NonZeroDurationError; + + fn try_from(duration_us: u64) -> Result { + IggyDuration::from(duration_us).try_into() + } +} + +impl From for IggyDuration { + fn from(duration: NonZeroIggyDuration) -> Self { + duration.duration + } +} + +impl FromStr for NonZeroIggyDuration { + type Err = NonZeroDurationError; + + fn from_str(s: &str) -> Result { + IggyDuration::from_str(s)?.try_into() + } +} + +impl Display for NonZeroIggyDuration { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.duration) + } +} + +impl Serialize for NonZeroIggyDuration { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + self.duration.serialize(serializer) + } +} + +struct NonZeroIggyDurationVisitor; + +impl<'de> Deserialize<'de> for NonZeroIggyDuration { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + deserializer.deserialize_u64(NonZeroIggyDurationVisitor) + } +} + +impl Visitor<'_> for NonZeroIggyDurationVisitor { + type Value = NonZeroIggyDuration; + + fn expecting(&self, formatter: &mut Formatter) -> std::fmt::Result { + formatter.write_str("a duration in microseconds greater than zero") + } + + fn visit_u64(self, value: u64) -> Result + where + E: DeError, + { + NonZeroIggyDuration::try_from(value).map_err(E::custom) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn given_a_positive_duration_should_convert() { + let duration = NonZeroIggyDuration::try_from(IggyDuration::ONE_SECOND).unwrap(); + + assert_eq!(IggyDuration::ONE_SECOND, duration.get()); + assert_eq!(Duration::from_secs(1), duration.get_duration()); + } + + #[test] + fn given_a_zero_duration_should_fail_to_convert() { + let error = NonZeroIggyDuration::try_from(IggyDuration::default()).unwrap_err(); + + assert_eq!(NonZeroDurationError::Zero, error); + } + + #[test] + fn given_a_zero_alias_should_fail_to_parse() { + for value in ["0", "0s", "none", "disabled", "unlimited"] { + assert_eq!( + Err(NonZeroDurationError::Zero), + NonZeroIggyDuration::from_str(value), + "expected {value} to be rejected" + ); + } + } + + #[test] + fn given_a_malformed_value_should_report_the_format_error() { + let error = NonZeroIggyDuration::from_str("1 hour and 30 minutes").unwrap_err(); + + assert!(matches!(error, NonZeroDurationError::InvalidFormat(_))); + } + + #[test] + fn given_a_human_time_string_should_parse() { + let duration = NonZeroIggyDuration::from_str("1h 1m 1s").unwrap(); + + assert_eq!(3661, duration.as_secs()); + assert_eq!("1h 1m 1s", duration.as_human_time_string()); + assert_eq!("1h 1m 1s", format!("{duration}")); + } + + #[test] + fn given_microseconds_should_round_trip_through_serde() { + let duration = NonZeroIggyDuration::from_str("500ms").unwrap(); + + let serialized = serde_json::to_string(&duration).unwrap(); + + assert_eq!("500000", serialized); + assert_eq!( + duration, + serde_json::from_str::(&serialized).unwrap() + ); + } + + #[test] + fn given_a_zero_microsecond_value_should_fail_to_deserialize() { + assert!(serde_json::from_str::("0").is_err()); + } +} From ca062b357e79836d67e393973bddc4aaba4392b8 Mon Sep 17 00:00:00 2001 From: Ethan Lin Date: Wed, 19 Aug 2026 23:35:54 +0800 Subject: [PATCH 2/4] refactor(sdk): type retry and heartbeat intervals as non-zero The transport heartbeat interval, the reconnection interval, the consumer init and polling retry intervals and the producer send retries interval all pace a loop, so zero either spins a core or panics tokio::time::interval. They now hold NonZeroIggyDuration, which rejects the zero where the config is built or the connection string is parsed. An auto-commit interval stays an IggyDuration: zero there stores the offsets in a busy loop, which is a supported choice. So does reestablish_after, where zero means reconnecting immediately. --- core/common/src/traits/binary_impls/system.rs | 6 +- core/common/src/traits/binary_transport.rs | 4 +- core/common/src/traits/system_client.rs | 6 +- .../auth_config/connection_string.rs | 8 +-- .../auth_config/connection_string_options.rs | 4 +- .../http_connection_string_options.rs | 34 ++++++++++-- .../quic_config/quic_client_config.rs | 6 +- .../quic_config/quic_client_config_builder.rs | 9 ++- .../quic_client_reconnection_config.rs | 9 +-- .../quic_connection_string_options.rs | 17 +++--- .../tcp_config/tcp_client_config.rs | 6 +- .../tcp_config/tcp_client_config_builder.rs | 7 ++- .../tcp_client_reconnection_config.rs | 7 ++- .../tcp_connection_string_options.rs | 17 +++--- .../websocket_client_config.rs | 6 +- .../websocket_client_config_builder.rs | 9 ++- .../websocket_client_reconnection_config.rs | 6 +- .../websocket_connection_string_options.rs | 17 +++--- ...ify_consumer_group_partition_assignment.rs | 4 +- .../server/scenarios/purge_delete_scenario.rs | 4 +- .../reconnect_after_restart_scenario.rs | 4 +- .../stale_client_consumer_group_scenario.rs | 8 +-- core/sdk/src/client_provider.rs | 38 ++++++++----- .../client_wrappers/binary_system_client.rs | 6 +- core/sdk/src/clients/binary_system.rs | 6 +- core/sdk/src/clients/client_builder.rs | 20 +++++-- core/sdk/src/clients/consumer.rs | 54 ++++++++++++++++-- core/sdk/src/clients/consumer_builder.rs | 16 +++--- core/sdk/src/clients/producer.rs | 8 +-- core/sdk/src/clients/producer_builder.rs | 9 +-- core/sdk/src/http/http_client.rs | 10 ++-- core/sdk/src/http/system.rs | 4 +- core/sdk/src/prelude.rs | 7 ++- core/sdk/src/quic/quic_client.rs | 36 +++++++++--- .../config/config_iggy_consumer.rs | 55 ++++++++++++------- .../config/config_iggy_producer.rs | 20 ++++--- core/sdk/src/tcp/tcp_client.rs | 48 ++++++++++++---- core/sdk/src/websocket/websocket_client.rs | 29 ++++++++-- .../stream-consumer-config/main.rs | 4 +- .../stream-producer-config/main.rs | 2 +- 40 files changed, 380 insertions(+), 190 deletions(-) diff --git a/core/common/src/traits/binary_impls/system.rs b/core/common/src/traits/binary_impls/system.rs index b9a526c09f..e3dc21db1a 100644 --- a/core/common/src/traits/binary_impls/system.rs +++ b/core/common/src/traits/binary_impls/system.rs @@ -18,8 +18,8 @@ use crate::traits::binary_auth::fail_if_not_authenticated; use crate::wire_conversions::clients_from_wire; use crate::{ - BinaryClient, ClientInfo, ClientInfoDetails, IggyDuration, IggyError, OptionSpec, OptionsScope, - Snapshot, SnapshotCompression, Stats, SystemClient, SystemSnapshotType, + BinaryClient, ClientInfo, ClientInfoDetails, IggyError, NonZeroIggyDuration, OptionSpec, + OptionsScope, Snapshot, SnapshotCompression, Stats, SystemClient, SystemSnapshotType, }; use iggy_binary_protocol::codec::WireEncode; use iggy_binary_protocol::codes::{ @@ -99,7 +99,7 @@ impl SystemClient for B { Ok(()) } - async fn heartbeat_interval(&self) -> IggyDuration { + async fn heartbeat_interval(&self) -> NonZeroIggyDuration { self.get_heartbeat_interval() } diff --git a/core/common/src/traits/binary_transport.rs b/core/common/src/traits/binary_transport.rs index d3c2bd2e3a..4c12db04bc 100644 --- a/core/common/src/traits/binary_transport.rs +++ b/core/common/src/traits/binary_transport.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use crate::{ClientState, DiagnosticEvent, IggyDuration, IggyError}; +use crate::{ClientState, DiagnosticEvent, IggyError, NonZeroIggyDuration}; use async_trait::async_trait; use bytes::Bytes; use std::sync::Arc; @@ -28,7 +28,7 @@ pub trait BinaryTransport { async fn set_state(&self, state: ClientState); async fn publish_event(&self, event: DiagnosticEvent); async fn send_raw_with_response(&self, code: u32, payload: Bytes) -> Result; - fn get_heartbeat_interval(&self) -> IggyDuration; + fn get_heartbeat_interval(&self) -> NonZeroIggyDuration; /// Per-transport consumer-group + partitioning cache used to resolve /// partitioning client-side under VSR (the broker never picks a diff --git a/core/common/src/traits/system_client.rs b/core/common/src/traits/system_client.rs index fe85ef9072..a418a5eb31 100644 --- a/core/common/src/traits/system_client.rs +++ b/core/common/src/traits/system_client.rs @@ -16,8 +16,8 @@ // under the License. use crate::{ - ClientInfo, ClientInfoDetails, IggyDuration, IggyError, OptionSpec, OptionsScope, Snapshot, - SnapshotCompression, Stats, SystemSnapshotType, + ClientInfo, ClientInfoDetails, IggyError, NonZeroIggyDuration, OptionSpec, OptionsScope, + Snapshot, SnapshotCompression, Stats, SystemSnapshotType, }; use async_trait::async_trait; @@ -49,7 +49,7 @@ pub trait SystemClient { async fn describe_options(&self, scope: OptionsScope) -> Result, IggyError>; /// Ping the server to check if it's alive. async fn ping(&self) -> Result<(), IggyError>; - async fn heartbeat_interval(&self) -> IggyDuration; + async fn heartbeat_interval(&self) -> NonZeroIggyDuration; /// Re-sync the cached consumer-group assignments from the coordinator. /// /// Driven off the heartbeat so a member picks up a new generation (e.g. a diff --git a/core/common/src/types/configuration/auth_config/connection_string.rs b/core/common/src/types/configuration/auth_config/connection_string.rs index 33c340177b..82691812f0 100644 --- a/core/common/src/types/configuration/auth_config/connection_string.rs +++ b/core/common/src/types/configuration/auth_config/connection_string.rs @@ -158,7 +158,7 @@ impl ConnectionStringUtils { #[cfg(test)] mod tests { use super::*; - use crate::IggyDuration; + use crate::NonZeroIggyDuration; use crate::TcpConnectionStringOptions; use secrecy::ExposeSecret; @@ -255,7 +255,7 @@ mod tests { assert!(connection_string.options.retries().is_none()); assert_eq!( connection_string.options.heartbeat_interval(), - IggyDuration::from_str("5s").unwrap() + NonZeroIggyDuration::from_str("5s").unwrap() ); } @@ -289,7 +289,7 @@ mod tests { assert_eq!(connection_string.options.retries().unwrap(), 3); assert_eq!( connection_string.options.heartbeat_interval(), - IggyDuration::from_str("10s").unwrap() + NonZeroIggyDuration::from_str("10s").unwrap() ); } @@ -317,7 +317,7 @@ mod tests { assert!(connection_string.options.retries().is_none()); assert_eq!( connection_string.options.heartbeat_interval(), - IggyDuration::from_str("5s").unwrap() + NonZeroIggyDuration::from_str("5s").unwrap() ); } } diff --git a/core/common/src/types/configuration/auth_config/connection_string_options.rs b/core/common/src/types/configuration/auth_config/connection_string_options.rs index 677b692e14..778a76e5eb 100644 --- a/core/common/src/types/configuration/auth_config/connection_string_options.rs +++ b/core/common/src/types/configuration/auth_config/connection_string_options.rs @@ -15,12 +15,12 @@ // specific language governing permissions and limitations // under the License. -use crate::{IggyDuration, IggyError}; +use crate::{IggyError, NonZeroIggyDuration}; pub trait ConnectionStringOptions { fn retries(&self) -> Option; - fn heartbeat_interval(&self) -> IggyDuration; + fn heartbeat_interval(&self) -> NonZeroIggyDuration; fn parse_options(options: &str) -> Result where diff --git a/core/common/src/types/configuration/http_config/http_connection_string_options.rs b/core/common/src/types/configuration/http_config/http_connection_string_options.rs index e101b35395..d75fa8bb65 100644 --- a/core/common/src/types/configuration/http_config/http_connection_string_options.rs +++ b/core/common/src/types/configuration/http_config/http_connection_string_options.rs @@ -15,12 +15,12 @@ // specific language governing permissions and limitations // under the License. -use crate::{ConnectionStringOptions, IggyDuration, IggyError}; +use crate::{ConnectionStringOptions, IggyError, NonZeroIggyDuration}; use std::str::FromStr; #[derive(Debug)] pub struct HttpConnectionStringOptions { - heartbeat_interval: IggyDuration, + heartbeat_interval: NonZeroIggyDuration, retries: u32, } @@ -29,7 +29,7 @@ impl ConnectionStringOptions for HttpConnectionStringOptions { Some(self.retries) } - fn heartbeat_interval(&self) -> IggyDuration { + fn heartbeat_interval(&self) -> NonZeroIggyDuration { self.heartbeat_interval } @@ -58,7 +58,7 @@ impl ConnectionStringOptions for HttpConnectionStringOptions { } } - let heartbeat_interval = IggyDuration::from_str(heartbeat_interval.as_str()) + let heartbeat_interval = NonZeroIggyDuration::from_str(heartbeat_interval.as_str()) .map_err(|_| IggyError::InvalidConnectionString)?; let connection_string_options = @@ -68,7 +68,7 @@ impl ConnectionStringOptions for HttpConnectionStringOptions { } impl HttpConnectionStringOptions { - pub fn new(heartbeat_interval: IggyDuration, retries: u32) -> Self { + pub fn new(heartbeat_interval: NonZeroIggyDuration, retries: u32) -> Self { Self { heartbeat_interval, retries, @@ -79,8 +79,30 @@ impl HttpConnectionStringOptions { impl Default for HttpConnectionStringOptions { fn default() -> Self { Self { - heartbeat_interval: IggyDuration::from_str("5s").unwrap(), + heartbeat_interval: NonZeroIggyDuration::from_str("5s").unwrap(), retries: 3, } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn should_parse_a_heartbeat_interval() { + let options = HttpConnectionStringOptions::parse_options("heartbeat_interval=10s").unwrap(); + + assert_eq!( + NonZeroIggyDuration::from_str("10s").unwrap(), + options.heartbeat_interval() + ); + } + + #[test] + fn should_fail_with_a_zero_heartbeat_interval() { + let error = HttpConnectionStringOptions::parse_options("heartbeat_interval=none").err(); + + assert!(matches!(error, Some(IggyError::InvalidConnectionString))); + } +} diff --git a/core/common/src/types/configuration/quic_config/quic_client_config.rs b/core/common/src/types/configuration/quic_config/quic_client_config.rs index b6e2b95e52..774924bb37 100644 --- a/core/common/src/types/configuration/quic_config/quic_client_config.rs +++ b/core/common/src/types/configuration/quic_config/quic_client_config.rs @@ -16,7 +16,7 @@ // under the License. use crate::{ - AutoLogin, ConnectionString, ConnectionStringOptions, IggyDuration, + AutoLogin, ConnectionString, ConnectionStringOptions, NonZeroIggyDuration, QuicClientReconnectionConfig, QuicConnectionStringOptions, }; use std::str::FromStr; @@ -53,7 +53,7 @@ pub struct QuicClientConfig { /// Whether to validate the server certificate. pub validate_certificate: bool, /// Interval of heartbeats sent by the client - pub heartbeat_interval: IggyDuration, + pub heartbeat_interval: NonZeroIggyDuration, } impl Default for QuicClientConfig { @@ -63,7 +63,7 @@ impl Default for QuicClientConfig { server_address: "127.0.0.1:8080".to_string(), server_name: "localhost".to_string(), auto_login: AutoLogin::Disabled, - heartbeat_interval: IggyDuration::from_str("5s").unwrap(), + heartbeat_interval: NonZeroIggyDuration::from_str("5s").unwrap(), reconnection: QuicClientReconnectionConfig::default(), response_buffer_size: 1000 * 1000 * 10, max_concurrent_bidi_streams: 10000, diff --git a/core/common/src/types/configuration/quic_config/quic_client_config_builder.rs b/core/common/src/types/configuration/quic_config/quic_client_config_builder.rs index 9af4e85024..0ade0cabb8 100644 --- a/core/common/src/types/configuration/quic_config/quic_client_config_builder.rs +++ b/core/common/src/types/configuration/quic_config/quic_client_config_builder.rs @@ -15,7 +15,10 @@ // specific language governing permissions and limitations // under the License. -use crate::{AutoLogin, IggyDuration, IggyError, QuicClientConfig, validate_server_address}; +use crate::{ + AutoLogin, IggyDuration, IggyError, NonZeroIggyDuration, QuicClientConfig, + validate_server_address, +}; /// Builder for the QUIC client configuration. /// @@ -81,7 +84,7 @@ impl QuicClientConfigBuilder { } /// Sets the interval between retries when connecting to the server. - pub fn with_reconnection_interval(mut self, interval: IggyDuration) -> Self { + pub fn with_reconnection_interval(mut self, interval: NonZeroIggyDuration) -> Self { self.config.reconnection.interval = interval; self } @@ -147,7 +150,7 @@ impl QuicClientConfigBuilder { } /// Sets the heartbeat interval. Defaults to 5000ms. - pub fn with_heartbeat_interval(mut self, interval: IggyDuration) -> Self { + pub fn with_heartbeat_interval(mut self, interval: NonZeroIggyDuration) -> Self { self.config.heartbeat_interval = interval; self } diff --git a/core/common/src/types/configuration/quic_config/quic_client_reconnection_config.rs b/core/common/src/types/configuration/quic_config/quic_client_reconnection_config.rs index 8191993e39..e5e6d4825d 100644 --- a/core/common/src/types/configuration/quic_config/quic_client_reconnection_config.rs +++ b/core/common/src/types/configuration/quic_config/quic_client_reconnection_config.rs @@ -15,14 +15,15 @@ // specific language governing permissions and limitations // under the License. -use crate::IggyDuration; +use crate::{IggyDuration, NonZeroIggyDuration}; use std::str::FromStr; #[derive(Debug, Clone)] pub struct QuicClientReconnectionConfig { pub enabled: bool, pub max_retries: Option, - pub interval: IggyDuration, + /// Delay between connection attempts. + pub interval: NonZeroIggyDuration, pub reestablish_after: IggyDuration, } @@ -30,7 +31,7 @@ impl QuicClientReconnectionConfig { pub fn new( enabled: bool, max_retries: Option, - interval: IggyDuration, + interval: NonZeroIggyDuration, reestablish_after: IggyDuration, ) -> Self { Self { @@ -47,7 +48,7 @@ impl Default for QuicClientReconnectionConfig { QuicClientReconnectionConfig { enabled: true, max_retries: None, - interval: IggyDuration::from_str("1s").unwrap(), + interval: NonZeroIggyDuration::from_str("1s").unwrap(), reestablish_after: IggyDuration::from_str("5s").unwrap(), } } diff --git a/core/common/src/types/configuration/quic_config/quic_connection_string_options.rs b/core/common/src/types/configuration/quic_config/quic_connection_string_options.rs index db937557e0..43f401661e 100644 --- a/core/common/src/types/configuration/quic_config/quic_connection_string_options.rs +++ b/core/common/src/types/configuration/quic_config/quic_connection_string_options.rs @@ -15,7 +15,10 @@ // specific language governing permissions and limitations // under the License. -use crate::{ConnectionStringOptions, IggyDuration, IggyError, QuicClientReconnectionConfig}; +use crate::{ + ConnectionStringOptions, IggyDuration, IggyError, NonZeroIggyDuration, + QuicClientReconnectionConfig, +}; use std::str::FromStr; #[derive(Debug)] @@ -30,7 +33,7 @@ pub struct QuicConnectionStringOptions { keep_alive_interval: u64, max_idle_timeout: u64, validate_certificate: bool, - heartbeat_interval: IggyDuration, + heartbeat_interval: NonZeroIggyDuration, } impl QuicConnectionStringOptions { @@ -80,7 +83,7 @@ impl ConnectionStringOptions for QuicConnectionStringOptions { self.reconnection.max_retries } - fn heartbeat_interval(&self) -> IggyDuration { + fn heartbeat_interval(&self) -> NonZeroIggyDuration { self.heartbeat_interval } @@ -203,13 +206,13 @@ impl ConnectionStringOptions for QuicConnectionStringOptions { .map_err(|_| IggyError::InvalidNumberValue)?, ), }, - interval: IggyDuration::from_str(reconnection_interval.as_str()) + interval: NonZeroIggyDuration::from_str(reconnection_interval.as_str()) .map_err(|_| IggyError::InvalidConnectionString)?, reestablish_after: IggyDuration::from_str(reconnection_reestablish_after.as_str()) .map_err(|_| IggyError::InvalidConnectionString)?, }; - let heartbeat_interval = IggyDuration::from_str(heartbeat_interval.as_str()) + let heartbeat_interval = NonZeroIggyDuration::from_str(heartbeat_interval.as_str()) .map_err(|_| IggyError::InvalidConnectionString)?; let connection_string_options = QuicConnectionStringOptions::new( @@ -243,7 +246,7 @@ impl QuicConnectionStringOptions { keep_alive_interval: u64, max_idle_timeout: u64, validate_certificate: bool, - heartbeat_interval: IggyDuration, + heartbeat_interval: NonZeroIggyDuration, ) -> Self { Self { reconnection, @@ -274,7 +277,7 @@ impl Default for QuicConnectionStringOptions { keep_alive_interval: 5000, max_idle_timeout: 10000, validate_certificate: false, - heartbeat_interval: IggyDuration::from_str("5s").unwrap(), + heartbeat_interval: NonZeroIggyDuration::from_str("5s").unwrap(), } } } diff --git a/core/common/src/types/configuration/tcp_config/tcp_client_config.rs b/core/common/src/types/configuration/tcp_config/tcp_client_config.rs index 86997c6d7a..b60f4b3ad0 100644 --- a/core/common/src/types/configuration/tcp_config/tcp_client_config.rs +++ b/core/common/src/types/configuration/tcp_config/tcp_client_config.rs @@ -18,7 +18,7 @@ use crate::types::configuration::auth_config::connection_string::ConnectionString; use crate::types::configuration::auth_config::connection_string_options::ConnectionStringOptions; use crate::types::configuration::tcp_config::tcp_connection_string_options::TcpConnectionStringOptions; -use crate::{AutoLogin, IggyDuration, TcpClientReconnectionConfig}; +use crate::{AutoLogin, NonZeroIggyDuration, TcpClientReconnectionConfig}; use std::str::FromStr; /// Configuration for the TCP client. @@ -40,7 +40,7 @@ pub struct TcpClientConfig { /// Whether to automatically reconnect when disconnected. pub reconnection: TcpClientReconnectionConfig, /// Interval of heartbeats sent by the client - pub heartbeat_interval: IggyDuration, + pub heartbeat_interval: NonZeroIggyDuration, /// Disable Nagle algorithm for the TCP socket. pub nodelay: bool, } @@ -53,7 +53,7 @@ impl Default for TcpClientConfig { tls_domain: "".to_string(), tls_ca_file: None, tls_validate_certificate: true, - heartbeat_interval: IggyDuration::from_str("5s").unwrap(), + heartbeat_interval: NonZeroIggyDuration::from_str("5s").unwrap(), auto_login: AutoLogin::Disabled, reconnection: TcpClientReconnectionConfig::default(), nodelay: false, diff --git a/core/common/src/types/configuration/tcp_config/tcp_client_config_builder.rs b/core/common/src/types/configuration/tcp_config/tcp_client_config_builder.rs index 6f665a5777..943b69f53e 100644 --- a/core/common/src/types/configuration/tcp_config/tcp_client_config_builder.rs +++ b/core/common/src/types/configuration/tcp_config/tcp_client_config_builder.rs @@ -15,7 +15,10 @@ // specific language governing permissions and limitations // under the License. -use crate::{AutoLogin, IggyDuration, IggyError, TcpClientConfig, validate_server_address}; +use crate::{ + AutoLogin, IggyDuration, IggyError, NonZeroIggyDuration, TcpClientConfig, + validate_server_address, +}; /// Builder for the TCP client configuration. /// Allows configuring the TCP client with custom settings or using defaults: @@ -59,7 +62,7 @@ impl TcpClientConfigBuilder { } /// Sets the interval between retries when connecting to the server. - pub fn with_reconnection_interval(mut self, interval: IggyDuration) -> Self { + pub fn with_reconnection_interval(mut self, interval: NonZeroIggyDuration) -> Self { self.config.reconnection.interval = interval; self } diff --git a/core/common/src/types/configuration/tcp_config/tcp_client_reconnection_config.rs b/core/common/src/types/configuration/tcp_config/tcp_client_reconnection_config.rs index 5a75184cfb..89d8644e8b 100644 --- a/core/common/src/types/configuration/tcp_config/tcp_client_reconnection_config.rs +++ b/core/common/src/types/configuration/tcp_config/tcp_client_reconnection_config.rs @@ -15,14 +15,15 @@ // specific language governing permissions and limitations // under the License. -use crate::IggyDuration; +use crate::{IggyDuration, NonZeroIggyDuration}; use std::str::FromStr; #[derive(Debug, Clone)] pub struct TcpClientReconnectionConfig { pub enabled: bool, pub max_retries: Option, - pub interval: IggyDuration, + /// Delay between connection attempts. + pub interval: NonZeroIggyDuration, pub reestablish_after: IggyDuration, } @@ -31,7 +32,7 @@ impl Default for TcpClientReconnectionConfig { TcpClientReconnectionConfig { enabled: true, max_retries: None, - interval: IggyDuration::from_str("1s").unwrap(), + interval: NonZeroIggyDuration::from_str("1s").unwrap(), reestablish_after: IggyDuration::from_str("5s").unwrap(), } } diff --git a/core/common/src/types/configuration/tcp_config/tcp_connection_string_options.rs b/core/common/src/types/configuration/tcp_config/tcp_connection_string_options.rs index 87ebd54490..1c957f1c37 100644 --- a/core/common/src/types/configuration/tcp_config/tcp_connection_string_options.rs +++ b/core/common/src/types/configuration/tcp_config/tcp_connection_string_options.rs @@ -15,7 +15,10 @@ // specific language governing permissions and limitations // under the License. -use crate::{ConnectionStringOptions, IggyDuration, IggyError, TcpClientReconnectionConfig}; +use crate::{ + ConnectionStringOptions, IggyDuration, IggyError, NonZeroIggyDuration, + TcpClientReconnectionConfig, +}; use std::str::FromStr; #[derive(Debug)] @@ -24,7 +27,7 @@ pub struct TcpConnectionStringOptions { tls_domain: String, tls_ca_file: Option, reconnection: TcpClientReconnectionConfig, - heartbeat_interval: IggyDuration, + heartbeat_interval: NonZeroIggyDuration, nodelay: bool, } @@ -55,7 +58,7 @@ impl ConnectionStringOptions for TcpConnectionStringOptions { self.reconnection.max_retries } - fn heartbeat_interval(&self) -> IggyDuration { + fn heartbeat_interval(&self) -> NonZeroIggyDuration { self.heartbeat_interval } @@ -116,13 +119,13 @@ impl ConnectionStringOptions for TcpConnectionStringOptions { .map_err(|_| IggyError::InvalidNumberValue)?, ), }, - interval: IggyDuration::from_str(reconnection_interval.as_str()) + interval: NonZeroIggyDuration::from_str(reconnection_interval.as_str()) .map_err(|_| IggyError::InvalidConnectionString)?, reestablish_after: IggyDuration::from_str(reestablish_after.as_str()) .map_err(|_| IggyError::InvalidConnectionString)?, }; - let heartbeat_interval = IggyDuration::from_str(heartbeat_interval.as_str()) + let heartbeat_interval = NonZeroIggyDuration::from_str(heartbeat_interval.as_str()) .map_err(|_| IggyError::InvalidConnectionString)?; let connection_string_options = TcpConnectionStringOptions::new( @@ -144,7 +147,7 @@ impl TcpConnectionStringOptions { tls_domain: String, tls_ca_file: Option, reconnection: TcpClientReconnectionConfig, - heartbeat_interval: IggyDuration, + heartbeat_interval: NonZeroIggyDuration, nodelay: bool, ) -> Self { Self { @@ -165,7 +168,7 @@ impl Default for TcpConnectionStringOptions { tls_domain: "".to_string(), tls_ca_file: None, reconnection: Default::default(), - heartbeat_interval: IggyDuration::from_str("5s").unwrap(), + heartbeat_interval: NonZeroIggyDuration::from_str("5s").unwrap(), nodelay: false, } } diff --git a/core/common/src/types/configuration/websocket_config/websocket_client_config.rs b/core/common/src/types/configuration/websocket_config/websocket_client_config.rs index d93c168a95..4295ab5270 100644 --- a/core/common/src/types/configuration/websocket_config/websocket_client_config.rs +++ b/core/common/src/types/configuration/websocket_config/websocket_client_config.rs @@ -17,7 +17,7 @@ use crate::types::configuration::auth_config::connection_string::ConnectionString; use crate::types::configuration::websocket_config::websocket_connection_string_options::WebSocketConnectionStringOptions; -use crate::{AutoLogin, IggyDuration, WebSocketClientReconnectionConfig}; +use crate::{AutoLogin, NonZeroIggyDuration, WebSocketClientReconnectionConfig}; use std::fmt::{Display, Formatter}; use std::str::FromStr; use tungstenite::protocol::WebSocketConfig as TungsteniteConfig; @@ -32,7 +32,7 @@ pub struct WebSocketClientConfig { /// Whether to automatically reconnect when disconnected. pub reconnection: WebSocketClientReconnectionConfig, /// Interval of heartbeats sent by the client - pub heartbeat_interval: IggyDuration, + pub heartbeat_interval: NonZeroIggyDuration, /// WebSocket-specific configuration. pub ws_config: WebSocketConfig, /// Whether tls is enabled @@ -69,7 +69,7 @@ impl Default for WebSocketClientConfig { server_address: "127.0.0.1:8092".to_string(), auto_login: AutoLogin::Disabled, reconnection: WebSocketClientReconnectionConfig::default(), - heartbeat_interval: IggyDuration::from_str("5s").unwrap(), + heartbeat_interval: NonZeroIggyDuration::from_str("5s").unwrap(), ws_config: WebSocketConfig::default(), tls_enabled: false, tls_domain: "localhost".to_string(), diff --git a/core/common/src/types/configuration/websocket_config/websocket_client_config_builder.rs b/core/common/src/types/configuration/websocket_config/websocket_client_config_builder.rs index 626c0e683e..89df1ba92c 100644 --- a/core/common/src/types/configuration/websocket_config/websocket_client_config_builder.rs +++ b/core/common/src/types/configuration/websocket_config/websocket_client_config_builder.rs @@ -15,7 +15,10 @@ // specific language governing permissions and limitations // under the License. -use crate::{AutoLogin, IggyDuration, IggyError, WebSocketClientConfig, validate_server_address}; +use crate::{ + AutoLogin, IggyDuration, IggyError, NonZeroIggyDuration, WebSocketClientConfig, + validate_server_address, +}; /// Builder for the WebSocket client configuration. /// Allows configuring the WebSocket client with custom settings or using defaults: @@ -58,7 +61,7 @@ impl WebSocketClientConfigBuilder { } /// Sets the interval between retries when connecting to the server. - pub fn with_reconnection_interval(mut self, interval: IggyDuration) -> Self { + pub fn with_reconnection_interval(mut self, interval: NonZeroIggyDuration) -> Self { self.config.reconnection.interval = interval; self } @@ -70,7 +73,7 @@ impl WebSocketClientConfigBuilder { } /// Sets the heartbeat interval. - pub fn with_heartbeat_interval(mut self, heartbeat_interval: IggyDuration) -> Self { + pub fn with_heartbeat_interval(mut self, heartbeat_interval: NonZeroIggyDuration) -> Self { self.config.heartbeat_interval = heartbeat_interval; self } diff --git a/core/common/src/types/configuration/websocket_config/websocket_client_reconnection_config.rs b/core/common/src/types/configuration/websocket_config/websocket_client_reconnection_config.rs index 5a4ad326c2..c1fed5e8e0 100644 --- a/core/common/src/types/configuration/websocket_config/websocket_client_reconnection_config.rs +++ b/core/common/src/types/configuration/websocket_config/websocket_client_reconnection_config.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use crate::IggyDuration; +use crate::{IggyDuration, NonZeroIggyDuration}; use serde::{Deserialize, Serialize}; use std::fmt::{Display, Formatter}; use std::str::FromStr; @@ -33,7 +33,7 @@ pub struct WebSocketClientReconnectionConfig { /// The maximum number of retries. If None, will retry infinitely. pub max_retries: Option, /// The interval between retries. - pub interval: IggyDuration, + pub interval: NonZeroIggyDuration, /// The time to wait before attempting to reestablish connection. pub reestablish_after: IggyDuration, } @@ -43,7 +43,7 @@ impl Default for WebSocketClientReconnectionConfig { WebSocketClientReconnectionConfig { enabled: true, max_retries: None, - interval: IggyDuration::from_str("1s").unwrap(), + interval: NonZeroIggyDuration::from_str("1s").unwrap(), reestablish_after: IggyDuration::from_str("5s").unwrap(), } } diff --git a/core/common/src/types/configuration/websocket_config/websocket_connection_string_options.rs b/core/common/src/types/configuration/websocket_config/websocket_connection_string_options.rs index 85c5fb2053..6bd0a4a331 100644 --- a/core/common/src/types/configuration/websocket_config/websocket_connection_string_options.rs +++ b/core/common/src/types/configuration/websocket_config/websocket_connection_string_options.rs @@ -15,12 +15,15 @@ // specific language governing permissions and limitations // under the License. -use crate::{ConnectionStringOptions, IggyDuration, IggyError, WebSocketClientReconnectionConfig}; +use crate::{ + ConnectionStringOptions, IggyDuration, IggyError, NonZeroIggyDuration, + WebSocketClientReconnectionConfig, +}; use std::str::FromStr; #[derive(Debug, Clone)] pub struct WebSocketConnectionStringOptions { - heartbeat_interval: IggyDuration, + heartbeat_interval: NonZeroIggyDuration, reconnection: WebSocketClientReconnectionConfig, read_buffer_size: Option, @@ -38,7 +41,7 @@ pub struct WebSocketConnectionStringOptions { } impl WebSocketConnectionStringOptions { - pub fn heartbeat_interval(&self) -> IggyDuration { + pub fn heartbeat_interval(&self) -> NonZeroIggyDuration { self.heartbeat_interval } @@ -92,7 +95,7 @@ impl ConnectionStringOptions for WebSocketConnectionStringOptions { self.reconnection.max_retries } - fn heartbeat_interval(&self) -> IggyDuration { + fn heartbeat_interval(&self) -> NonZeroIggyDuration { self.heartbeat_interval } @@ -111,7 +114,7 @@ impl ConnectionStringOptions for WebSocketConnectionStringOptions { match parts[0] { "heartbeat_interval" => { - parsed_options.heartbeat_interval = IggyDuration::from_str(parts[1]) + parsed_options.heartbeat_interval = NonZeroIggyDuration::from_str(parts[1]) .map_err(|_| IggyError::InvalidConnectionString)?; } "reconnection_retries" => { @@ -125,7 +128,7 @@ impl ConnectionStringOptions for WebSocketConnectionStringOptions { parsed_options.reconnection.max_retries = retries; } "reconnection_interval" => { - parsed_options.reconnection.interval = IggyDuration::from_str(parts[1]) + parsed_options.reconnection.interval = NonZeroIggyDuration::from_str(parts[1]) .map_err(|_| IggyError::InvalidConnectionString)?; } "reestablish_after" => { @@ -192,7 +195,7 @@ impl ConnectionStringOptions for WebSocketConnectionStringOptions { impl Default for WebSocketConnectionStringOptions { fn default() -> Self { WebSocketConnectionStringOptions { - heartbeat_interval: IggyDuration::from_str("5s").unwrap(), + heartbeat_interval: NonZeroIggyDuration::from_str("5s").unwrap(), reconnection: WebSocketClientReconnectionConfig::default(), read_buffer_size: None, write_buffer_size: None, diff --git a/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs b/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs index d392755aea..5eafdbed4f 100644 --- a/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs +++ b/core/integration/tests/data_integrity/verify_consumer_group_partition_assignment.rs @@ -60,7 +60,7 @@ const CONSUMER1_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(1); async fn create_stale_tcp_client(server_addr: &str) -> IggyClient { let config = TcpClientConfig { server_address: server_addr.to_string(), - heartbeat_interval: IggyDuration::from_str("1h").unwrap(), + heartbeat_interval: NonZeroIggyDuration::from_str("1h").unwrap(), nodelay: true, ..TcpClientConfig::default() }; @@ -72,7 +72,7 @@ async fn create_stale_tcp_client(server_addr: &str) -> IggyClient { async fn create_tcp_client(server_addr: &str) -> IggyClient { let config = TcpClientConfig { server_address: server_addr.to_string(), - heartbeat_interval: IggyDuration::from_str("500ms").unwrap(), + heartbeat_interval: NonZeroIggyDuration::from_str("500ms").unwrap(), nodelay: true, ..TcpClientConfig::default() }; diff --git a/core/integration/tests/server/scenarios/purge_delete_scenario.rs b/core/integration/tests/server/scenarios/purge_delete_scenario.rs index a217e1f261..ce222ec623 100644 --- a/core/integration/tests/server/scenarios/purge_delete_scenario.rs +++ b/core/integration/tests/server/scenarios/purge_delete_scenario.rs @@ -1342,7 +1342,7 @@ async fn maybe_restart(harness: &mut TestHarness, client: &IggyClient, restart_s /// credentials in the transport config so the SDK re-authenticates on reconnect. fn build_root_client(harness: &TestHarness) -> IggyClient { let addr = harness.server().tcp_addr().unwrap(); - let interval = IggyDuration::from_str("200ms").unwrap(); + let interval = NonZeroIggyDuration::from_str("200ms").unwrap(); IggyClient::builder() .with_tcp() .with_server_address(addr.to_string()) @@ -1352,7 +1352,7 @@ fn build_root_client(harness: &TestHarness) -> IggyClient { ))) .with_reconnection_max_retries(Some(10)) .with_reconnection_interval(interval) - .with_reestablish_after(interval) + .with_reestablish_after(interval.get()) .build() .unwrap() } diff --git a/core/integration/tests/server/scenarios/reconnect_after_restart_scenario.rs b/core/integration/tests/server/scenarios/reconnect_after_restart_scenario.rs index 0ce68b8507..b7b48d3ff9 100644 --- a/core/integration/tests/server/scenarios/reconnect_after_restart_scenario.rs +++ b/core/integration/tests/server/scenarios/reconnect_after_restart_scenario.rs @@ -47,7 +47,7 @@ pub async fn run_producer(harness: &mut TestHarness) { .expect("Failed to create producer builder") .create_stream_if_not_exists() .create_topic_if_not_exists(1, IggyExpiry::NeverExpire, MaxTopicSize::ServerDefault) - .send_retries(Some(10), Some(IggyDuration::from_str("2s").unwrap())) + .send_retries(Some(10), Some(NonZeroIggyDuration::from_str("2s").unwrap())) .build(); producer @@ -142,7 +142,7 @@ pub async fn run_consumer(harness: &mut TestHarness) { .polling_strategy(PollingStrategy::next()) .batch_length(10) .poll_interval(IggyDuration::from_str("100ms").unwrap()) - .polling_retry_interval(IggyDuration::from_str("500ms").unwrap()) + .polling_retry_interval(NonZeroIggyDuration::from_str("500ms").unwrap()) .build(); consumer diff --git a/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs b/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs index 40772c0907..ec130dc660 100644 --- a/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs +++ b/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs @@ -40,7 +40,7 @@ const TOTAL_MESSAGES: u32 = 10; async fn create_client(server_addr: &str, heartbeat_interval: &str) -> IggyClient { let config = TcpClientConfig { server_address: server_addr.to_string(), - heartbeat_interval: IggyDuration::from_str(heartbeat_interval).unwrap(), + heartbeat_interval: NonZeroIggyDuration::from_str(heartbeat_interval).unwrap(), nodelay: true, ..TcpClientConfig::default() }; @@ -52,7 +52,7 @@ async fn create_client(server_addr: &str, heartbeat_interval: &str) -> IggyClien async fn create_reconnecting_client(server_addr: &str) -> IggyClient { let config = TcpClientConfig { server_address: server_addr.to_string(), - heartbeat_interval: IggyDuration::from_str("1h").unwrap(), + heartbeat_interval: NonZeroIggyDuration::from_str("1h").unwrap(), nodelay: true, auto_login: AutoLogin::Enabled(Credentials::UsernamePassword( DEFAULT_ROOT_USERNAME.to_string(), @@ -61,7 +61,7 @@ async fn create_reconnecting_client(server_addr: &str) -> IggyClient { reconnection: TcpClientReconnectionConfig { enabled: true, max_retries: Some(5), - interval: IggyDuration::from_str("500ms").unwrap(), + interval: NonZeroIggyDuration::from_str("500ms").unwrap(), reestablish_after: IggyDuration::from_str("100ms").unwrap(), }, ..TcpClientConfig::default() @@ -284,7 +284,7 @@ async fn should_handle_stale_client_with_auto_reconnection( .auto_join_consumer_group() .create_consumer_group_if_not_exists() .auto_commit(AutoCommit::When(AutoCommitWhen::PollingMessages)) - .polling_retry_interval(IggyDuration::from_str("500ms").unwrap()) + .polling_retry_interval(NonZeroIggyDuration::from_str("500ms").unwrap()) .build(); consumer.init().await.unwrap(); diff --git a/core/sdk/src/client_provider.rs b/core/sdk/src/client_provider.rs index 423b7048ea..b4c28303a1 100644 --- a/core/sdk/src/client_provider.rs +++ b/core/sdk/src/client_provider.rs @@ -19,8 +19,8 @@ use crate::client_wrappers::client_wrapper::ClientWrapper; use crate::clients::client::IggyClient; use crate::http::http_client::HttpClient; use crate::prelude::{ - ClientError, HttpClientConfig, IggyDuration, QuicClientConfig, QuicClientReconnectionConfig, - TcpClientConfig, TcpClientReconnectionConfig, WebSocketClient, + ClientError, HttpClientConfig, IggyDuration, IggyError, NonZeroIggyDuration, QuicClientConfig, + QuicClientReconnectionConfig, TcpClientConfig, TcpClientReconnectionConfig, WebSocketClient, }; use crate::quic::quic_client::QuicClient; use crate::tcp::tcp_client::TcpClient; @@ -92,16 +92,19 @@ impl ClientProviderConfig { client_address: args.quic_client_address, server_address: args.quic_server_address, server_name: args.quic_server_name, - heartbeat_interval: IggyDuration::from_str(&args.quic_heartbeat_interval) - .unwrap(), + heartbeat_interval: NonZeroIggyDuration::from_str( + &args.quic_heartbeat_interval, + ) + .map_err(|_| IggyError::InvalidConfiguration)?, reconnection: QuicClientReconnectionConfig { enabled: args.quic_reconnection_enabled, max_retries: args.quic_reconnection_max_retries, - interval: IggyDuration::from_str(&args.quic_reconnection_interval).unwrap(), + interval: NonZeroIggyDuration::from_str(&args.quic_reconnection_interval) + .map_err(|_| IggyError::InvalidConfiguration)?, reestablish_after: IggyDuration::from_str( &args.quic_reconnection_reestablish_after, ) - .unwrap(), + .map_err(|_| IggyError::InvalidConfiguration)?, }, auto_login: if auto_login { AutoLogin::Enabled(Credentials::UsernamePassword( @@ -137,16 +140,17 @@ impl ClientProviderConfig { tls_ca_file: args.tcp_tls_ca_file, tls_validate_certificate: true, nodelay: args.tcp_nodelay, - heartbeat_interval: IggyDuration::from_str(&args.tcp_heartbeat_interval) - .unwrap(), + heartbeat_interval: NonZeroIggyDuration::from_str(&args.tcp_heartbeat_interval) + .map_err(|_| IggyError::InvalidConfiguration)?, reconnection: TcpClientReconnectionConfig { enabled: args.tcp_reconnection_enabled, max_retries: args.tcp_reconnection_max_retries, - interval: IggyDuration::from_str(&args.tcp_reconnection_interval).unwrap(), + interval: NonZeroIggyDuration::from_str(&args.tcp_reconnection_interval) + .map_err(|_| IggyError::InvalidConfiguration)?, reestablish_after: IggyDuration::from_str( &args.tcp_reconnection_reestablish_after, ) - .unwrap(), + .map_err(|_| IggyError::InvalidConfiguration)?, }, auto_login: if auto_login { AutoLogin::Enabled(Credentials::UsernamePassword( @@ -161,17 +165,21 @@ impl ClientProviderConfig { TransportProtocol::WebSocket => { config.websocket = Some(Arc::new(WebSocketClientConfig { server_address: args.websocket_server_address, - heartbeat_interval: IggyDuration::from_str(&args.websocket_heartbeat_interval) - .unwrap(), + heartbeat_interval: NonZeroIggyDuration::from_str( + &args.websocket_heartbeat_interval, + ) + .map_err(|_| IggyError::InvalidConfiguration)?, reconnection: WebSocketClientReconnectionConfig { enabled: args.websocket_reconnection_enabled, max_retries: args.websocket_reconnection_max_retries, - interval: IggyDuration::from_str(&args.websocket_reconnection_interval) - .unwrap(), + interval: NonZeroIggyDuration::from_str( + &args.websocket_reconnection_interval, + ) + .map_err(|_| IggyError::InvalidConfiguration)?, reestablish_after: IggyDuration::from_str( &args.websocket_reconnection_reestablish_after, ) - .unwrap(), + .map_err(|_| IggyError::InvalidConfiguration)?, }, auto_login: if auto_login { AutoLogin::Enabled(Credentials::UsernamePassword( diff --git a/core/sdk/src/client_wrappers/binary_system_client.rs b/core/sdk/src/client_wrappers/binary_system_client.rs index e474377492..19c89cae5e 100644 --- a/core/sdk/src/client_wrappers/binary_system_client.rs +++ b/core/sdk/src/client_wrappers/binary_system_client.rs @@ -19,8 +19,8 @@ use crate::client_wrappers::client_wrapper::ClientWrapper; use async_trait::async_trait; use iggy_common::SystemClient; use iggy_common::{ - ClientInfo, ClientInfoDetails, IggyDuration, IggyError, OptionSpec, OptionsScope, Snapshot, - SnapshotCompression, Stats, SystemSnapshotType, + ClientInfo, ClientInfoDetails, IggyError, NonZeroIggyDuration, OptionSpec, OptionsScope, + Snapshot, SnapshotCompression, Stats, SystemSnapshotType, }; #[async_trait] @@ -85,7 +85,7 @@ impl SystemClient for ClientWrapper { } } - async fn heartbeat_interval(&self) -> IggyDuration { + async fn heartbeat_interval(&self) -> NonZeroIggyDuration { match self { ClientWrapper::Iggy(client) => client.heartbeat_interval().await, ClientWrapper::Http(client) => client.heartbeat_interval().await, diff --git a/core/sdk/src/clients/binary_system.rs b/core/sdk/src/clients/binary_system.rs index a2681696f7..7d5d07c1cd 100644 --- a/core/sdk/src/clients/binary_system.rs +++ b/core/sdk/src/clients/binary_system.rs @@ -20,8 +20,8 @@ use async_trait::async_trait; use iggy_common::SystemClient; use iggy_common::locking::IggyRwLockFn; use iggy_common::{ - ClientInfo, ClientInfoDetails, IggyDuration, IggyError, OptionSpec, OptionsScope, Snapshot, - SnapshotCompression, Stats, SystemSnapshotType, + ClientInfo, ClientInfoDetails, IggyError, NonZeroIggyDuration, OptionSpec, OptionsScope, + Snapshot, SnapshotCompression, Stats, SystemSnapshotType, }; #[async_trait] @@ -50,7 +50,7 @@ impl SystemClient for IggyClient { self.client.read().await.ping().await } - async fn heartbeat_interval(&self) -> IggyDuration { + async fn heartbeat_interval(&self) -> NonZeroIggyDuration { self.client.read().await.heartbeat_interval().await } diff --git a/core/sdk/src/clients/client_builder.rs b/core/sdk/src/clients/client_builder.rs index 0af473b554..20a960c6d3 100644 --- a/core/sdk/src/clients/client_builder.rs +++ b/core/sdk/src/clients/client_builder.rs @@ -19,8 +19,9 @@ use crate::client_wrappers::client_wrapper::ClientWrapper; use crate::clients::client::IggyClient; use crate::http::http_client::HttpClient; use crate::prelude::{ - AutoLogin, EncryptorKind, HttpClientConfigBuilder, IggyDuration, IggyError, Partitioner, - QuicClientConfigBuilder, TcpClientConfigBuilder, WebSocketClientConfigBuilder, + AutoLogin, EncryptorKind, HttpClientConfigBuilder, IggyDuration, IggyError, + NonZeroIggyDuration, Partitioner, QuicClientConfigBuilder, TcpClientConfigBuilder, + WebSocketClientConfigBuilder, }; use crate::quic::quic_client::QuicClient; use crate::tcp::tcp_client::TcpClient; @@ -174,7 +175,10 @@ impl TcpClientBuilder { } /// Sets the interval between retries when connecting to the server. - pub fn with_reconnection_interval(mut self, reconnection_interval: IggyDuration) -> Self { + pub fn with_reconnection_interval( + mut self, + reconnection_interval: NonZeroIggyDuration, + ) -> Self { self.config = self .config .with_reconnection_interval(reconnection_interval); @@ -258,7 +262,10 @@ impl QuicClientBuilder { } /// Sets the interval between retries when connecting to the server. - pub fn with_reconnection_interval(mut self, reconnection_interval: IggyDuration) -> Self { + pub fn with_reconnection_interval( + mut self, + reconnection_interval: NonZeroIggyDuration, + ) -> Self { self.config = self .config .with_reconnection_interval(reconnection_interval); @@ -351,7 +358,10 @@ impl WebSocketClientBuilder { } /// Sets the interval between retries when connecting to the server. - pub fn with_reconnection_interval(mut self, reconnection_interval: IggyDuration) -> Self { + pub fn with_reconnection_interval( + mut self, + reconnection_interval: NonZeroIggyDuration, + ) -> Self { self.config = self .config .with_reconnection_interval(reconnection_interval); diff --git a/core/sdk/src/clients/consumer.rs b/core/sdk/src/clients/consumer.rs index a935ca24de..4b00c51112 100644 --- a/core/sdk/src/clients/consumer.rs +++ b/core/sdk/src/clients/consumer.rs @@ -26,7 +26,8 @@ use iggy_common::{ }; use iggy_common::{ Consumer, ConsumerKind, DiagnosticEvent, EncryptorKind, IdKind, Identifier, IggyDuration, - IggyError, IggyMessage, IggyTimestamp, PolledMessages, PollingKind, PollingStrategy, + IggyError, IggyMessage, IggyTimestamp, NonZeroIggyDuration, PolledMessages, PollingKind, + PollingStrategy, }; use std::collections::VecDeque; use std::future::Future; @@ -50,6 +51,7 @@ pub enum AutoCommit { /// The auto-commit is disabled and the offset must be stored manually by the consumer. Disabled, /// The auto-commit is enabled and the offset is stored on the server after a certain interval. + /// A zero interval stores the offsets in a busy loop. Interval(IggyDuration), /// The auto-commit is enabled and the offset is stored on the server after a certain interval or depending on the mode when consuming the messages. IntervalOrWhen(IggyDuration, AutoCommitWhen), @@ -134,9 +136,9 @@ pub struct IggyConsumer { store_after_every_nth_message: u64, last_polled_at: Arc, current_partition_id: Arc, - reconnection_retry_interval: IggyDuration, + reconnection_retry_interval: NonZeroIggyDuration, init_retries: Option, - init_retry_interval: IggyDuration, + init_retry_interval: NonZeroIggyDuration, allow_replay: bool, offset_drain_timeout: IggyDuration, } @@ -157,9 +159,9 @@ impl IggyConsumer { auto_join_consumer_group: bool, create_consumer_group_if_not_exists: bool, encryptor: Option>, - reconnection_retry_interval: IggyDuration, + reconnection_retry_interval: NonZeroIggyDuration, init_retries: Option, - init_retry_interval: IggyDuration, + init_retry_interval: NonZeroIggyDuration, allow_replay: bool, offset_drain_timeout: IggyDuration, ) -> Self { @@ -1239,3 +1241,45 @@ impl Drop for IggyConsumer { ); } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::client_wrappers::client_wrapper::ClientWrapper; + use crate::clients::consumer_builder::IggyConsumerBuilder; + use crate::tcp::tcp_client::TcpClient; + use iggy_common::locking::IggyRwLockFn; + + fn builder() -> IggyConsumerBuilder { + IggyConsumerBuilder::new( + IggyRwLock::new(ClientWrapper::Tcp(TcpClient::default())), + "consumer".to_owned(), + Consumer::new(Identifier::numeric(1).unwrap()), + Identifier::numeric(1).unwrap(), + Identifier::numeric(1).unwrap(), + None, + None, + None, + ) + } + + #[tokio::test] + async fn should_accept_auto_commit_modes_with_a_zero_interval() { + let zero = IggyDuration::new(Duration::ZERO); + + for auto_commit in [ + AutoCommit::Interval(zero), + AutoCommit::IntervalOrWhen(zero, AutoCommitWhen::PollingMessages), + AutoCommit::IntervalOrAfter(zero, AutoCommitAfter::ConsumingAllMessages), + ] { + let mut consumer = builder().auto_commit(auto_commit).build(); + + let error = consumer.init().await.err(); + + assert!( + !matches!(error, Some(IggyError::InvalidConfiguration)), + "{auto_commit:?} must be accepted" + ); + } + } +} diff --git a/core/sdk/src/clients/consumer_builder.rs b/core/sdk/src/clients/consumer_builder.rs index 50ff187869..49d50faf23 100644 --- a/core/sdk/src/clients/consumer_builder.rs +++ b/core/sdk/src/clients/consumer_builder.rs @@ -18,7 +18,9 @@ use crate::client_wrappers::client_wrapper::ClientWrapper; use crate::prelude::{AutoCommit, AutoCommitWhen, IggyConsumer}; use iggy_common::locking::IggyRwLock; -use iggy_common::{Consumer, EncryptorKind, Identifier, IggyDuration, PollingStrategy}; +use iggy_common::{ + Consumer, EncryptorKind, Identifier, IggyDuration, NonZeroIggyDuration, PollingStrategy, +}; use std::sync::Arc; #[derive(Debug)] @@ -36,9 +38,9 @@ pub struct IggyConsumerBuilder { auto_join_consumer_group: bool, create_consumer_group_if_not_exists: bool, encryptor: Option>, - polling_retry_interval: IggyDuration, + polling_retry_interval: NonZeroIggyDuration, init_retries: Option, - init_retry_interval: IggyDuration, + init_retry_interval: NonZeroIggyDuration, allow_replay: bool, offset_drain_timeout: IggyDuration, } @@ -72,9 +74,9 @@ impl IggyConsumerBuilder { create_consumer_group_if_not_exists: true, encryptor, polling_interval, - polling_retry_interval: IggyDuration::ONE_SECOND, + polling_retry_interval: NonZeroIggyDuration::ONE_SECOND, init_retries: None, - init_retry_interval: IggyDuration::ONE_SECOND, + init_retry_interval: NonZeroIggyDuration::ONE_SECOND, allow_replay: false, offset_drain_timeout: IggyDuration::new_from_secs(5), } @@ -191,7 +193,7 @@ impl IggyConsumerBuilder { } /// Sets the polling retry interval in case of server disconnection. - pub fn polling_retry_interval(self, interval: IggyDuration) -> Self { + pub fn polling_retry_interval(self, interval: NonZeroIggyDuration) -> Self { Self { polling_retry_interval: interval, ..self @@ -201,7 +203,7 @@ impl IggyConsumerBuilder { /// Sets the number of retries and the interval when initializing the consumer if the stream or topic is not found. /// Might be useful when the stream or topic is created dynamically by the producer. /// By default, the consumer will not retry. - pub fn init_retries(self, retries: u32, interval: IggyDuration) -> Self { + pub fn init_retries(self, retries: u32, interval: NonZeroIggyDuration) -> Self { Self { init_retries: Some(retries), init_retry_interval: interval, diff --git a/core/sdk/src/clients/producer.rs b/core/sdk/src/clients/producer.rs index 04912b1daa..a6c3e57918 100644 --- a/core/sdk/src/clients/producer.rs +++ b/core/sdk/src/clients/producer.rs @@ -26,8 +26,8 @@ use futures_util::StreamExt; use iggy_common::locking::{IggyRwLock, IggyRwLockFn}; use iggy_common::{Client, MessageClient, StreamClient, TopicClient, TopicCreateOptions}; use iggy_common::{ - DiagnosticEvent, EncryptorKind, IdKind, Identifier, IggyDuration, IggyError, IggyExpiry, - IggyMessage, IggyTimestamp, MaxTopicSize, Partitioner, Partitioning, + DiagnosticEvent, EncryptorKind, IdKind, Identifier, IggyError, IggyExpiry, IggyMessage, + IggyTimestamp, MaxTopicSize, NonZeroIggyDuration, Partitioner, Partitioning, SendMessagesConfirmationResponse, SendMessagesResponse, }; use std::sync::Arc; @@ -99,7 +99,7 @@ pub struct ProducerCore { default_partitioning: Arc, last_sent_at: Arc, send_retries_count: Option, - send_retries_interval: Option, + send_retries_interval: Option, direct_config: Option, } @@ -496,7 +496,7 @@ impl IggyProducer { topic_message_expiry: IggyExpiry, topic_max_size: MaxTopicSize, send_retries_count: Option, - send_retries_interval: Option, + send_retries_interval: Option, mode: SendMode, ) -> Self { let core = Arc::new(ProducerCore { diff --git a/core/sdk/src/clients/producer_builder.rs b/core/sdk/src/clients/producer_builder.rs index 7734e2e778..0d05adbcb5 100644 --- a/core/sdk/src/clients/producer_builder.rs +++ b/core/sdk/src/clients/producer_builder.rs @@ -20,7 +20,8 @@ use crate::clients::producer_config::{BackgroundConfig, DirectConfig}; use crate::prelude::IggyProducer; use iggy_common::locking::IggyRwLock; use iggy_common::{ - EncryptorKind, Identifier, IggyDuration, IggyExpiry, MaxTopicSize, Partitioner, Partitioning, + EncryptorKind, Identifier, IggyExpiry, MaxTopicSize, NonZeroIggyDuration, Partitioner, + Partitioning, }; use std::sync::Arc; @@ -47,7 +48,7 @@ pub struct IggyProducerBuilder { create_topic_if_not_exists: bool, topic_partitions_count: u32, send_retries_count: Option, - send_retries_interval: Option, + send_retries_interval: Option, topic_message_expiry: IggyExpiry, topic_max_size: MaxTopicSize, partitioning: Option, @@ -80,7 +81,7 @@ impl IggyProducerBuilder { topic_message_expiry: IggyExpiry::ServerDefault, topic_max_size: MaxTopicSize::ServerDefault, send_retries_count: Some(3), - send_retries_interval: Some(IggyDuration::ONE_SECOND), + send_retries_interval: Some(NonZeroIggyDuration::ONE_SECOND), mode: SendMode::default(), } } @@ -186,7 +187,7 @@ impl IggyProducerBuilder { /// Sets the retry policy (maximum number of retries and interval between them) in case of messages sending failure. /// The error can be related either to disconnecting from the server or to the server rejecting the messages. /// Default is 3 retries with 1 second interval between them. - pub fn send_retries(self, retries: Option, interval: Option) -> Self { + pub fn send_retries(self, retries: Option, interval: Option) -> Self { Self { send_retries_count: retries, send_retries_interval: interval, diff --git a/core/sdk/src/http/http_client.rs b/core/sdk/src/http/http_client.rs index 7ad6652890..20f4888319 100644 --- a/core/sdk/src/http/http_client.rs +++ b/core/sdk/src/http/http_client.rs @@ -16,7 +16,7 @@ // under the License. use crate::http::http_transport::HttpTransport; -use crate::prelude::{Client, HttpClientConfig, IggyDuration, IggyError}; +use crate::prelude::{Client, HttpClientConfig, IggyError, NonZeroIggyDuration}; use async_broadcast::{Receiver, Sender, broadcast}; use async_trait::async_trait; use bytes::Bytes; @@ -48,7 +48,7 @@ const PUBLIC_PATHS: &[&str] = &[ pub struct HttpClient { /// The URL of the Iggy API. pub api_url: Url, - pub(crate) heartbeat_interval: IggyDuration, + pub(crate) heartbeat_interval: NonZeroIggyDuration, client: ClientWithMiddleware, access_token: IggyRwLock, events: (Sender, Receiver), @@ -279,7 +279,7 @@ impl HttpClient { Ok(Self { api_url, client, - heartbeat_interval: IggyDuration::from_str("5s").unwrap(), + heartbeat_interval: NonZeroIggyDuration::from_str("5s").unwrap(), access_token: IggyRwLock::new(access_token), events: broadcast(1000), }) @@ -528,7 +528,7 @@ mod tests { ); assert_eq!( http_client.as_ref().unwrap().heartbeat_interval, - IggyDuration::from_str("5s").unwrap() + NonZeroIggyDuration::from_str("5s").unwrap() ); } @@ -570,7 +570,7 @@ mod tests { ); assert_eq!( http_client.as_ref().unwrap().heartbeat_interval, - IggyDuration::from_str("5s").unwrap() + NonZeroIggyDuration::from_str("5s").unwrap() ); } diff --git a/core/sdk/src/http/system.rs b/core/sdk/src/http/system.rs index 4634c9daf5..09d2e9f8b0 100644 --- a/core/sdk/src/http/system.rs +++ b/core/sdk/src/http/system.rs @@ -17,7 +17,7 @@ use crate::http::http_client::HttpClient; use crate::http::http_transport::HttpTransport; -use crate::prelude::{IggyDuration, IggyError}; +use crate::prelude::{IggyError, NonZeroIggyDuration}; use async_trait::async_trait; use iggy_common::Snapshot; use iggy_common::Stats; @@ -86,7 +86,7 @@ impl SystemClient for HttpClient { Ok(()) } - async fn heartbeat_interval(&self) -> IggyDuration { + async fn heartbeat_interval(&self) -> NonZeroIggyDuration { self.heartbeat_interval } diff --git a/core/sdk/src/prelude.rs b/core/sdk/src/prelude.rs index f3d4db5d78..326432ce3f 100644 --- a/core/sdk/src/prelude.rs +++ b/core/sdk/src/prelude.rs @@ -55,9 +55,10 @@ pub use iggy_common::{ HeaderKind, HeaderValue, HttpClientConfig, HttpClientConfigBuilder, HttpMethod, IdKind, Identifier, IdentityInfo, IggyByteSize, IggyDuration, IggyError, IggyExpiry, IggyIndexView, IggyMessage, IggyMessageHeader, IggyMessageHeaderView, IggyMessageView, - IggyMessageViewIterator, IggyTimestamp, MaxTopicSize, OptionSpec, OptionValue, OptionsScope, - Partition, Partitioner, Partitioning, Permissions, PersonalAccessTokenExpiry, PollMessages, - PolledMessages, PollingKind, PollingStrategy, QuicClientConfig, QuicClientConfigBuilder, + IggyMessageViewIterator, IggyTimestamp, MaxTopicSize, NonZeroDurationError, + NonZeroIggyDuration, OptionSpec, OptionValue, OptionsScope, Partition, Partitioner, + Partitioning, Permissions, PersonalAccessTokenExpiry, PollMessages, PolledMessages, + PollingKind, PollingStrategy, QuicClientConfig, QuicClientConfigBuilder, QuicClientReconnectionConfig, ResourceOptions, SendMessages, SendMessagesConfirmationResponse, SendMessagesResponse, Sizeable, SnapshotCompression, Stats, Stream, StreamDetails, StreamPermissions, StreamUpdateOptions, SystemSnapshotType, TcpClientConfig, diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs index 921ef5d3dc..b253205548 100644 --- a/core/sdk/src/quic/quic_client.rs +++ b/core/sdk/src/quic/quic_client.rs @@ -23,7 +23,9 @@ use crate::session::ConsensusSession; use iggy_common::VsrSessionControl as _; use iggy_common::{BinaryClient, BinaryTransport, Client, PersonalAccessTokenClient, UserClient}; -use crate::prelude::{IggyDuration, IggyError, IggyTimestamp, QuicClientConfig}; +use crate::prelude::{ + IggyDuration, IggyError, IggyTimestamp, NonZeroIggyDuration, QuicClientConfig, +}; use crate::quic::skip_server_verification::SkipServerVerification; use async_broadcast::{Receiver, Sender, broadcast}; use async_trait::async_trait; @@ -182,7 +184,7 @@ impl BinaryTransport for QuicClient { self.send_raw(code, payload).await } - fn get_heartbeat_interval(&self) -> IggyDuration { + fn get_heartbeat_interval(&self) -> NonZeroIggyDuration { self.config.heartbeat_interval } @@ -781,6 +783,24 @@ fn configure(config: &QuicClientConfig) -> Result { mod tests { use super::*; + #[tokio::test] + async fn should_fail_with_a_zero_heartbeat_interval() { + let value = "iggy+quic://user:secret@127.0.0.1:1234?heartbeat_interval=none"; + + let error = QuicClient::from_connection_string(value).err(); + + assert!(matches!(error, Some(IggyError::InvalidConnectionString))); + } + + #[tokio::test] + async fn should_fail_with_a_zero_reconnection_interval() { + let value = "iggy+quic://user:secret@127.0.0.1:1234?reconnection_interval=0"; + + let error = QuicClient::from_connection_string(value).err(); + + assert!(matches!(error, Some(IggyError::InvalidConnectionString))); + } + #[tokio::test] async fn should_fail_with_empty_connection_string() { let value = ""; @@ -945,14 +965,14 @@ mod tests { assert!(!quic_client_config.validate_certificate); assert_eq!( quic_client_config.heartbeat_interval, - IggyDuration::from_str("5s").unwrap() + NonZeroIggyDuration::from_str("5s").unwrap() ); assert!(quic_client_config.reconnection.enabled); assert!(quic_client_config.reconnection.max_retries.is_none()); assert_eq!( quic_client_config.reconnection.interval, - IggyDuration::from_str("1s").unwrap() + NonZeroIggyDuration::from_str("1s").unwrap() ); assert_eq!( quic_client_config.reconnection.reestablish_after, @@ -1003,14 +1023,14 @@ mod tests { assert!(!quic_client_config.validate_certificate); assert_eq!( quic_client_config.heartbeat_interval, - IggyDuration::from_str("5s").unwrap() + NonZeroIggyDuration::from_str("5s").unwrap() ); assert!(quic_client_config.reconnection.enabled); assert!(quic_client_config.reconnection.max_retries.is_none()); assert_eq!( quic_client_config.reconnection.interval, - IggyDuration::from_str(reconnection_interval).unwrap() + NonZeroIggyDuration::from_str(reconnection_interval).unwrap() ); assert_eq!( quic_client_config.reconnection.reestablish_after, @@ -1052,14 +1072,14 @@ mod tests { assert!(!quic_client_config.validate_certificate); assert_eq!( quic_client_config.heartbeat_interval, - IggyDuration::from_str("5s").unwrap() + NonZeroIggyDuration::from_str("5s").unwrap() ); assert!(quic_client_config.reconnection.enabled); assert!(quic_client_config.reconnection.max_retries.is_none()); assert_eq!( quic_client_config.reconnection.interval, - IggyDuration::from_str("1s").unwrap() + NonZeroIggyDuration::from_str("1s").unwrap() ); assert_eq!( quic_client_config.reconnection.reestablish_after, diff --git a/core/sdk/src/stream_builder/config/config_iggy_consumer.rs b/core/sdk/src/stream_builder/config/config_iggy_consumer.rs index 4f872b20fb..ccac7344f6 100644 --- a/core/sdk/src/stream_builder/config/config_iggy_consumer.rs +++ b/core/sdk/src/stream_builder/config/config_iggy_consumer.rs @@ -17,7 +17,8 @@ use crate::clients::consumer::{AutoCommit, AutoCommitWhen}; use crate::prelude::{ - ConsumerKind, EncryptorKind, Identifier, IggyDuration, IggyError, PollingStrategy, + ConsumerKind, EncryptorKind, Identifier, IggyDuration, IggyError, NonZeroIggyDuration, + PollingStrategy, }; use bon::Builder; use std::str::FromStr; @@ -55,11 +56,11 @@ pub struct IggyConsumerConfig { /// `PollingStrategy` specifies from where to start polling messages. See `PollingStrategy` for details. polling_strategy: PollingStrategy, /// Sets the polling retry interval in case of server disconnection. - polling_retry_interval: IggyDuration, + polling_retry_interval: NonZeroIggyDuration, /// Sets the number of retries and the interval when initializing the consumer if the stream or topic is not found. /// Might be useful when the stream or topic is created dynamically by the producer. init_retries: Option, - init_interval: IggyDuration, + init_interval: NonZeroIggyDuration, /// Sets a optional client side encryptor for encrypting the messages' payloads. Currently only Aes256Gcm is supported. /// Note, this is independent of server side encryption meaning you can add client encryption, server encryption, or both. encryptor: Option>, @@ -85,9 +86,9 @@ impl Default for IggyConsumerConfig { polling_strategy: PollingStrategy::last(), partitions_count: 1, encryptor: None, - polling_retry_interval: IggyDuration::new_from_secs(1), + polling_retry_interval: NonZeroIggyDuration::ONE_SECOND, init_retries: Some(5), - init_interval: IggyDuration::new_from_secs(3), + init_interval: NonZeroIggyDuration::from_str("3s").unwrap(), } } } @@ -135,9 +136,9 @@ impl IggyConsumerConfig { polling_strategy: PollingStrategy, partitions_count: u32, encryptor: Option>, - polling_retry_interval: IggyDuration, + polling_retry_interval: NonZeroIggyDuration, init_retries: Option, - init_interval: IggyDuration, + init_interval: NonZeroIggyDuration, ) -> Self { Self { stream_id, @@ -196,9 +197,9 @@ impl IggyConsumerConfig { polling_strategy: PollingStrategy::last(), partitions_count: 1, encryptor: None, - polling_retry_interval: IggyDuration::new_from_secs(1), + polling_retry_interval: NonZeroIggyDuration::ONE_SECOND, init_retries: Some(5), - init_interval: IggyDuration::new_from_secs(3), + init_interval: NonZeroIggyDuration::from_str("3s").unwrap(), }) } } @@ -259,7 +260,7 @@ impl IggyConsumerConfig { self.encryptor.clone() } - pub fn polling_retry_interval(&self) -> IggyDuration { + pub fn polling_retry_interval(&self) -> NonZeroIggyDuration { self.polling_retry_interval } @@ -267,7 +268,7 @@ impl IggyConsumerConfig { self.init_retries } - pub fn init_interval(&self) -> IggyDuration { + pub fn init_interval(&self) -> NonZeroIggyDuration { self.init_interval } } @@ -295,10 +296,10 @@ mod tests { .consumer_kind(ConsumerKind::ConsumerGroup) .polling_interval(IggyDuration::from_str("5ms").unwrap()) .polling_strategy(PollingStrategy::last()) - .polling_retry_interval(IggyDuration::new_from_secs(1)) + .polling_retry_interval(NonZeroIggyDuration::ONE_SECOND) .partitions_count(1) .init_retries(3) - .init_interval(IggyDuration::new_from_secs(3)) + .init_interval(NonZeroIggyDuration::from_str("3s").unwrap()) .build(); assert_eq!( @@ -329,11 +330,14 @@ mod tests { assert_eq!( config.polling_retry_interval(), - IggyDuration::new_from_secs(1) + NonZeroIggyDuration::ONE_SECOND ); assert_eq!(config.init_retries(), Some(3)); - assert_eq!(config.init_interval(), IggyDuration::new_from_secs(3)); + assert_eq!( + config.init_interval(), + NonZeroIggyDuration::from_str("3s").unwrap() + ); } #[test] @@ -362,9 +366,15 @@ mod tests { assert_eq!(config.polling_strategy(), PollingStrategy::last()); assert_eq!(config.partitions_count(), 1); - assert_eq!(config.polling_retry_interval(), IggyDuration::ONE_SECOND); + assert_eq!( + config.polling_retry_interval(), + NonZeroIggyDuration::ONE_SECOND + ); assert_eq!(config.init_retries(), Some(5)); - assert_eq!(config.init_interval(), IggyDuration::new_from_secs(3)); + assert_eq!( + config.init_interval(), + NonZeroIggyDuration::from_str("3s").unwrap() + ); } #[test] @@ -384,9 +394,9 @@ mod tests { PollingStrategy::last(), 1, None, - IggyDuration::new_from_secs(1), + NonZeroIggyDuration::ONE_SECOND, Some(3), - IggyDuration::new_from_secs(3), + NonZeroIggyDuration::from_str("3s").unwrap(), ); assert_eq!( config.stream_id(), @@ -416,10 +426,13 @@ mod tests { assert_eq!( config.polling_retry_interval(), - IggyDuration::new_from_secs(1) + NonZeroIggyDuration::ONE_SECOND ); assert_eq!(config.init_retries(), Some(3)); - assert_eq!(config.init_interval(), IggyDuration::new_from_secs(3)); + assert_eq!( + config.init_interval(), + NonZeroIggyDuration::from_str("3s").unwrap() + ); } #[test] diff --git a/core/sdk/src/stream_builder/config/config_iggy_producer.rs b/core/sdk/src/stream_builder/config/config_iggy_producer.rs index 19dc4cea7b..311448e428 100644 --- a/core/sdk/src/stream_builder/config/config_iggy_producer.rs +++ b/core/sdk/src/stream_builder/config/config_iggy_producer.rs @@ -15,7 +15,9 @@ // specific language governing permissions and limitations // under the License. -use crate::prelude::{EncryptorKind, Identifier, IggyDuration, IggyError, Partitioning}; +use crate::prelude::{ + EncryptorKind, Identifier, IggyDuration, IggyError, NonZeroIggyDuration, Partitioning, +}; use bon::Builder; use std::str::FromStr; use std::sync::Arc; @@ -43,7 +45,7 @@ pub struct IggyProducerConfig { /// Sets the maximum number of send retries in case of a message sending failure. send_retries_count: Option, /// Sets the interval between send retries in case of a message sending failure. - send_retries_interval: Option, + send_retries_interval: Option, /// Sets a optional client side encryptor for encrypting the messages' payloads. Currently only Aes256Gcm is supported. /// Note, this is independent of server side encryption meaning you can add client encryption, server encryption, or both. encryptor: Option>, @@ -65,7 +67,7 @@ impl Default for IggyProducerConfig { topic_partitions_count: 1, encryptor: None, send_retries_count: Some(3), - send_retries_interval: Some(IggyDuration::new_from_secs(1)), + send_retries_interval: Some(NonZeroIggyDuration::ONE_SECOND), } } } @@ -102,7 +104,7 @@ impl IggyProducerConfig { partitioning: Partitioning, encryptor: Option>, send_retries_count: Option, - send_retries_interval: Option, + send_retries_interval: Option, ) -> Self { Self { stream_id, @@ -152,7 +154,7 @@ impl IggyProducerConfig { topic_partitions_count: 1, encryptor: None, send_retries_count: Some(3), - send_retries_interval: Some(IggyDuration::new_from_secs(1)), + send_retries_interval: Some(NonZeroIggyDuration::ONE_SECOND), }) } } @@ -198,7 +200,7 @@ impl IggyProducerConfig { self.send_retries_count } - pub fn send_retries_interval(&self) -> Option { + pub fn send_retries_interval(&self) -> Option { self.send_retries_interval } } @@ -223,7 +225,7 @@ mod tests { .linger_time(IggyDuration::from_str("5ms").unwrap()) .partitioning(Partitioning::balanced()) .send_retries_count(3) - .send_retries_interval(IggyDuration::new_from_secs(1)) + .send_retries_interval(NonZeroIggyDuration::ONE_SECOND) .build(); assert_eq!( @@ -243,7 +245,7 @@ mod tests { assert_eq!(config.send_retries_count(), Some(3)); assert_eq!( config.send_retries_interval(), - Some(IggyDuration::new_from_secs(1)) + Some(NonZeroIggyDuration::ONE_SECOND) ); } @@ -264,7 +266,7 @@ mod tests { assert_eq!(config.send_retries_count(), Some(3)); assert_eq!( config.send_retries_interval(), - Some(IggyDuration::new_from_secs(1)) + Some(NonZeroIggyDuration::ONE_SECOND) ); } diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs index 2c09e3f2ef..1edd3a029f 100644 --- a/core/sdk/src/tcp/tcp_client.rs +++ b/core/sdk/src/tcp/tcp_client.rs @@ -31,7 +31,8 @@ use iggy_binary_protocol::codes::{LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_C use iggy_common::VsrSessionControl as _; use iggy_common::{ AutoLogin, ClientState, ConnectionString, ConnectionStringUtils, Credentials, DiagnosticEvent, - IggyDuration, IggyError, IggyTimestamp, TcpConnectionStringOptions, TransportProtocol, + IggyDuration, IggyError, IggyTimestamp, NonZeroIggyDuration, TcpConnectionStringOptions, + TransportProtocol, }; use iggy_common::{BinaryClient, BinaryTransport, PersonalAccessTokenClient, UserClient}; use rustls::pki_types::{CertificateDer, ServerName, pem::PemObject}; @@ -192,7 +193,7 @@ impl BinaryTransport for TcpClient { self.send_raw(code, payload).await } - fn get_heartbeat_interval(&self) -> IggyDuration { + fn get_heartbeat_interval(&self) -> NonZeroIggyDuration { self.config.heartbeat_interval } @@ -242,7 +243,7 @@ impl TcpClient { pub fn new( server_address: &str, auto_sign_in: AutoLogin, - heartbeat_interval: IggyDuration, + heartbeat_interval: NonZeroIggyDuration, ) -> Result { Self::create(Arc::new(TcpClientConfig { heartbeat_interval, @@ -257,7 +258,7 @@ impl TcpClient { server_address: &str, domain: &str, auto_sign_in: AutoLogin, - heartbeat_interval: IggyDuration, + heartbeat_interval: NonZeroIggyDuration, ) -> Result { Self::create(Arc::new(TcpClientConfig { heartbeat_interval, @@ -871,6 +872,33 @@ const fn is_login_register_code(code: u32) -> bool { mod tests { use super::*; + #[test] + fn should_fail_with_a_zero_heartbeat_interval() { + let value = "iggy+tcp://user:secret@127.0.0.1:1234?heartbeat_interval=none"; + + let error = TcpClient::from_connection_string(value).err(); + + assert!(matches!(error, Some(IggyError::InvalidConnectionString))); + } + + #[test] + fn should_succeed_with_a_zero_reestablish_after() { + let value = "iggy+tcp://user:secret@127.0.0.1:1234?reestablish_after=0"; + + let client = TcpClient::from_connection_string(value).unwrap(); + + assert!(client.config.reconnection.reestablish_after.is_zero()); + } + + #[test] + fn should_fail_with_a_zero_reconnection_interval() { + let value = "iggy+tcp://user:secret@127.0.0.1:1234?reconnection_interval=0"; + + let error = TcpClient::from_connection_string(value).err(); + + assert!(matches!(error, Some(IggyError::InvalidConnectionString))); + } + #[test] fn should_fail_with_empty_connection_string() { let value = ""; @@ -1029,14 +1057,14 @@ mod tests { assert!(tcp_client_config.tls_ca_file.is_none()); assert_eq!( tcp_client_config.heartbeat_interval, - IggyDuration::from_str("5s").unwrap() + NonZeroIggyDuration::from_str("5s").unwrap() ); assert!(tcp_client_config.reconnection.enabled); assert!(tcp_client_config.reconnection.max_retries.is_none()); assert_eq!( tcp_client_config.reconnection.interval, - IggyDuration::from_str("1s").unwrap() + NonZeroIggyDuration::from_str("1s").unwrap() ); assert_eq!( tcp_client_config.reconnection.reestablish_after, @@ -1078,7 +1106,7 @@ mod tests { assert!(tcp_client_config.tls_ca_file.is_none()); assert_eq!( tcp_client_config.heartbeat_interval, - IggyDuration::from_str(heartbeat_interval).unwrap() + NonZeroIggyDuration::from_str(heartbeat_interval).unwrap() ); assert!(tcp_client_config.reconnection.enabled); @@ -1088,7 +1116,7 @@ mod tests { ); assert_eq!( tcp_client_config.reconnection.interval, - IggyDuration::from_str("1s").unwrap() + NonZeroIggyDuration::from_str("1s").unwrap() ); assert_eq!( tcp_client_config.reconnection.reestablish_after, @@ -1124,14 +1152,14 @@ mod tests { assert!(tcp_client_config.tls_ca_file.is_none()); assert_eq!( tcp_client_config.heartbeat_interval, - IggyDuration::from_str("5s").unwrap() + NonZeroIggyDuration::from_str("5s").unwrap() ); assert!(tcp_client_config.reconnection.enabled); assert!(tcp_client_config.reconnection.max_retries.is_none()); assert_eq!( tcp_client_config.reconnection.interval, - IggyDuration::from_str("1s").unwrap() + NonZeroIggyDuration::from_str("1s").unwrap() ); assert_eq!( tcp_client_config.reconnection.reestablish_after, diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs index c439a38f19..47f7f4ac6a 100644 --- a/core/sdk/src/websocket/websocket_client.rs +++ b/core/sdk/src/websocket/websocket_client.rs @@ -32,7 +32,8 @@ use iggy_binary_protocol::codes::{LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_C use iggy_common::VsrSessionControl as _; use iggy_common::{ AutoLogin, ClientState, ConnectionString, Credentials, DiagnosticEvent, IggyDuration, - IggyError, IggyTimestamp, WebSocketClientConfig, WebSocketConnectionStringOptions, + IggyError, IggyTimestamp, NonZeroIggyDuration, WebSocketClientConfig, + WebSocketConnectionStringOptions, }; use iggy_common::{BinaryClient, BinaryTransport, PersonalAccessTokenClient, UserClient}; use secrecy::ExposeSecret; @@ -178,7 +179,7 @@ impl BinaryTransport for WebSocketClient { self.send_raw(code, payload).await } - fn get_heartbeat_interval(&self) -> IggyDuration { + fn get_heartbeat_interval(&self) -> NonZeroIggyDuration { self.config.heartbeat_interval } @@ -763,7 +764,7 @@ mod tests { assert_eq!(client.config.server_address, "127.0.0.1:8092"); assert_eq!( client.config.heartbeat_interval, - IggyDuration::from_str("5s").unwrap() + NonZeroIggyDuration::from_str("5s").unwrap() ); assert!(matches!(client.config.auto_login, AutoLogin::Disabled)); assert!(client.config.reconnection.enabled); @@ -786,7 +787,7 @@ mod tests { fn should_create_with_custom_config() { let config = WebSocketClientConfig { server_address: "localhost:9090".to_string(), - heartbeat_interval: IggyDuration::from_str("10s").unwrap(), + heartbeat_interval: NonZeroIggyDuration::from_str("10s").unwrap(), ..Default::default() }; @@ -797,10 +798,28 @@ mod tests { assert_eq!(client.config.server_address, "localhost:9090"); assert_eq!( client.config.heartbeat_interval, - IggyDuration::from_str("10s").unwrap() + NonZeroIggyDuration::from_str("10s").unwrap() ); } + #[test] + fn should_fail_with_a_zero_heartbeat_interval() { + let value = "iggy+ws://user:secret@127.0.0.1:1234?heartbeat_interval=none"; + + let error = WebSocketClient::from_connection_string(value).err(); + + assert!(matches!(error, Some(IggyError::InvalidConnectionString))); + } + + #[test] + fn should_fail_with_a_zero_reconnection_interval() { + let value = "iggy+ws://user:secret@127.0.0.1:1234?reconnection_interval=0"; + + let error = WebSocketClient::from_connection_string(value).err(); + + assert!(matches!(error, Some(IggyError::InvalidConnectionString))); + } + #[test] fn should_fail_with_empty_connection_string() { let value = ""; diff --git a/examples/rust/src/stream-builder/stream-consumer-config/main.rs b/examples/rust/src/stream-builder/stream-consumer-config/main.rs index 6a3cd62255..7e7b886b39 100644 --- a/examples/rust/src/stream-builder/stream-consumer-config/main.rs +++ b/examples/rust/src/stream-builder/stream-consumer-config/main.rs @@ -69,14 +69,14 @@ async fn main() -> Result<(), IggyError> { // - `Next` - start polling from the next message after the last polled message based on the stored consumer offset. .polling_strategy(PollingStrategy::last()) // Sets the polling retry interval in case of server disconnection. - .polling_retry_interval(IggyDuration::new_from_secs(1)) + .polling_retry_interval(NonZeroIggyDuration::ONE_SECOND) // Sets the number of retries and the interval when initializing the consumer if the stream or topic is not found. // Might be useful when the stream or topic is created dynamically by the producer. // The retry only occurs when configured and is disabled by default. // When you want to retry at most 5 times with an interval of 1 second, // you set `init_retries` to 5 and `init_interval` to 1 second. .init_retries(5) - .init_interval(IggyDuration::new_from_secs(1)) + .init_interval(NonZeroIggyDuration::ONE_SECOND) // Optionally, set a custom client side encryptor for encrypting the messages' payloads. Currently only Aes256Gcm is supported. // Key must be identical to the one used by the producer; thus ensure secure key exchange i.e. K8s secret etc. // Note, this is independent of server side encryption meaning you can add client encryption, server encryption, or both. diff --git a/examples/rust/src/stream-builder/stream-producer-config/main.rs b/examples/rust/src/stream-builder/stream-producer-config/main.rs index 5f7c9a2263..af7783bd62 100644 --- a/examples/rust/src/stream-builder/stream-producer-config/main.rs +++ b/examples/rust/src/stream-builder/stream-producer-config/main.rs @@ -53,7 +53,7 @@ async fn main() -> Result<(), IggyError> { // The error can be related either to disconnecting from the server or to the server rejecting the messages. // Default is 3 retries with 1 second interval between them. Customize to your requirements. .send_retries_count(3) - .send_retries_interval(IggyDuration::new_from_secs(1)) + .send_retries_interval(NonZeroIggyDuration::ONE_SECOND) // Optionally, set a custom client side encryptor for encrypting the messages' payloads. Currently only Aes256Gcm is supported. // Note, this is independent of server side encryption meaning you can add client encryption, server encryption, or both. // .encryptor( Arc::new(EncryptorKind::Aes256Gcm(Aes256GcmEncryptor::new(&[1; 32])?))) From 35484683b816b67cdc96e7a608c1d637f5eea3ed Mon Sep 17 00:00:00 2001 From: Ethan Lin Date: Wed, 19 Aug 2026 23:35:58 +0800 Subject: [PATCH 3/4] refactor(python): reject zero retry and heartbeat intervals unconditionally The binding follows the Rust SDK: a zero reconnection interval is now rejected whatever the retry policy is, and a zero auto-commit interval is accepted and stores the offsets in a busy loop. --- foreign/python/apache_iggy.pyi | 7 +- foreign/python/src/client.rs | 4 +- foreign/python/src/config.rs | 21 ++---- foreign/python/src/consumer.rs | 18 +++-- foreign/python/src/duration.rs | 16 ++--- foreign/python/tests/test_client_config.py | 79 +++++++++++++--------- 6 files changed, 73 insertions(+), 72 deletions(-) diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 50b65786db..a5b9fc39ce 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -1345,8 +1345,8 @@ class IggyClient: Creates a new consumer group consumer. Returns the consumer or a RuntimeError on failure. Raises `ValueError` if `poll_interval`, `polling_retry_interval`, `init_retry_interval` or an - `AutoCommit` interval is negative, or if any of those except `poll_interval` - is zero. + `AutoCommit` interval is negative, or if `polling_retry_interval` or + `init_retry_interval` is zero. """ def send_binary_request( self, code: builtins.int, payload: builtins.bytes @@ -1989,8 +1989,7 @@ class TcpReconnectionConfig: Raises: ValueError: If a duration is negative, if `max_retries` is outside the - range of an unsigned 32-bit integer, or if `interval` is zero while - reconnection is enabled and `max_retries` is unlimited. + range of an unsigned 32-bit integer, or if `interval` is zero. """ def __repr__(self) -> builtins.str: ... diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index 816ba4e61f..a8da9e8c3a 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -1084,8 +1084,8 @@ impl IggyClient { /// Creates a new consumer group consumer. /// Returns the consumer or a RuntimeError on failure. Raises `ValueError` if /// `poll_interval`, `polling_retry_interval`, `init_retry_interval` or an - /// `AutoCommit` interval is negative, or if any of those except `poll_interval` - /// is zero. + /// `AutoCommit` interval is negative, or if `polling_retry_interval` or + /// `init_retry_interval` is zero. #[allow(clippy::too_many_arguments)] #[pyo3(signature = ( name, diff --git a/foreign/python/src/config.rs b/foreign/python/src/config.rs index 4519c939fb..79eb285b16 100644 --- a/foreign/python/src/config.rs +++ b/foreign/python/src/config.rs @@ -132,8 +132,7 @@ impl TcpReconnectionConfig { /// /// Raises: /// ValueError: If a duration is negative, if `max_retries` is outside the - /// range of an unsigned 32-bit integer, or if `interval` is zero while - /// reconnection is enabled and `max_retries` is unlimited. + /// range of an unsigned 32-bit integer, or if `interval` is zero. #[new] #[pyo3(signature = (*, enabled=None, max_retries=None, interval=None, reestablish_after=None))] fn new( @@ -160,15 +159,9 @@ impl TcpReconnectionConfig { .as_ref() .map(py_delta_to_iggy_duration) .transpose()? + .map(|interval| reject_zero(interval, "interval")) + .transpose()? .unwrap_or(defaults.interval); - // Unlimited retries at a zero interval reconnect in a continuous loop; - // a zero interval with a retry cap is a legitimate fast-retry policy, and - // with reconnection off the interval is never read at all. - if enabled && interval.is_zero() && max_retries.is_none() { - return Err(PyValueError::new_err( - "'interval' must not be zero unless 'max_retries' is set", - )); - } Ok(Self { inner: RustTcpClientReconnectionConfig { enabled, @@ -197,7 +190,7 @@ impl TcpReconnectionConfig { #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] #[getter] fn interval<'a>(&self, py: Python<'a>) -> PyResult> { - iggy_duration_to_py_delta(py, self.inner.interval) + iggy_duration_to_py_delta(py, self.inner.interval.get()) } #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] @@ -214,7 +207,7 @@ impl TcpReconnectionConfig { format!( "TcpReconnectionConfig(enabled={}, max_retries={max_retries}, interval={}, reestablish_after={})", python_bool(self.inner.enabled), - duration_repr(self.inner.interval), + duration_repr(self.inner.interval.get()), duration_repr(self.inner.reestablish_after), ) } @@ -360,7 +353,7 @@ impl TcpConfig { #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] #[getter] fn heartbeat_interval<'a>(&self, py: Python<'a>) -> PyResult> { - iggy_duration_to_py_delta(py, self.inner.heartbeat_interval) + iggy_duration_to_py_delta(py, self.inner.heartbeat_interval.get()) } #[getter] @@ -399,7 +392,7 @@ impl TcpConfig { self.inner.server_address, self.auto_login().__repr__(), self.reconnection().__repr__(), - duration_repr(self.inner.heartbeat_interval), + duration_repr(self.inner.heartbeat_interval.get()), python_bool(self.inner.tls_enabled), self.inner.tls_domain, python_bool(self.inner.tls_validate_certificate), diff --git a/foreign/python/src/consumer.rs b/foreign/python/src/consumer.rs index 6a6e69a877..918df9a1d4 100644 --- a/foreign/python/src/consumer.rs +++ b/foreign/python/src/consumer.rs @@ -23,8 +23,8 @@ use iggy::prelude::{ AutoCommit as RustAutoCommit, AutoCommitAfter as RustAutoCommitAfter, AutoCommitWhen as RustAutoCommitWhen, ConsumerGroup as RustConsumerGroup, ConsumerGroupDetails as RustConsumerGroupDetails, - ConsumerGroupMember as RustConsumerGroupMember, IggyConsumer as RustIggyConsumer, IggyDuration, - IggyError, ReceivedMessage, + ConsumerGroupMember as RustConsumerGroupMember, IggyConsumer as RustIggyConsumer, IggyError, + ReceivedMessage, }; use pyo3::exceptions::PyStopAsyncIteration; use pyo3::types::PyDelta; @@ -38,7 +38,7 @@ use tokio::sync::Mutex; use tokio::sync::oneshot::Sender; use tokio::task::JoinHandle; -use crate::duration::{py_delta_to_iggy_duration, reject_zero}; +use crate::duration::py_delta_to_iggy_duration; use crate::identifier::PyIdentifier; use crate::receive_message::ReceiveMessage; @@ -432,12 +432,14 @@ impl TryFrom<&AutoCommit> for RustAutoCommit { fn try_from(val: &AutoCommit) -> PyResult { Ok(match val { AutoCommit::Disabled() => RustAutoCommit::Disabled, - AutoCommit::Interval(delta) => RustAutoCommit::Interval(auto_commit_interval(delta)?), + AutoCommit::Interval(delta) => { + RustAutoCommit::Interval(py_delta_to_iggy_duration(delta)?) + } AutoCommit::IntervalOrWhen(delta, when) => { - RustAutoCommit::IntervalOrWhen(auto_commit_interval(delta)?, when.into()) + RustAutoCommit::IntervalOrWhen(py_delta_to_iggy_duration(delta)?, when.into()) } AutoCommit::IntervalOrAfter(delta, after) => { - RustAutoCommit::IntervalOrAfter(auto_commit_interval(delta)?, after.into()) + RustAutoCommit::IntervalOrAfter(py_delta_to_iggy_duration(delta)?, after.into()) } AutoCommit::When(when) => RustAutoCommit::When(when.into()), AutoCommit::After(after) => RustAutoCommit::After(after.into()), @@ -445,10 +447,6 @@ impl TryFrom<&AutoCommit> for RustAutoCommit { } } -fn auto_commit_interval(delta: &Py) -> PyResult { - reject_zero(py_delta_to_iggy_duration(delta)?, "AutoCommit interval") -} - /// The auto-commit mode for storing the offset on the server. #[derive(Debug, PartialEq, Copy, Clone)] #[gen_stub_pyclass_complex_enum(skip_stub_type)] diff --git a/foreign/python/src/duration.rs b/foreign/python/src/duration.rs index 9b2b419fe5..1d448ad254 100644 --- a/foreign/python/src/duration.rs +++ b/foreign/python/src/duration.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use iggy::prelude::IggyDuration; +use iggy::prelude::{IggyDuration, NonZeroIggyDuration}; use pyo3::exceptions::PyValueError; use pyo3::prelude::*; use pyo3::types::PyDelta; @@ -53,13 +53,9 @@ pub fn duration_repr(duration: IggyDuration) -> String { } } -/// Rejects a zero duration for parameters where zero means an unthrottled loop -/// rather than "disabled". -pub fn reject_zero(duration: IggyDuration, parameter: &str) -> PyResult { - if duration.is_zero() { - return Err(PyValueError::new_err(format!( - "'{parameter}' must not be zero" - ))); - } - Ok(duration) +/// Converts a duration for parameters that pace a loop, where zero means an +/// unthrottled loop rather than "disabled". +pub fn reject_zero(duration: IggyDuration, parameter: &str) -> PyResult { + NonZeroIggyDuration::try_from(duration) + .map_err(|_| PyValueError::new_err(format!("'{parameter}' must not be zero"))) } diff --git a/foreign/python/tests/test_client_config.py b/foreign/python/tests/test_client_config.py index 78df23459c..39237a355b 100644 --- a/foreign/python/tests/test_client_config.py +++ b/foreign/python/tests/test_client_config.py @@ -143,22 +143,23 @@ def test_zero_reestablish_after_is_allowed(self): assert reconnection.reestablish_after == timedelta(0) - def test_zero_interval_is_allowed_with_bounded_retries(self): - """Test that a zero interval is legal as a bounded fast-retry policy.""" - reconnection = TcpReconnectionConfig(interval=timedelta(0), max_retries=5) - - assert reconnection.interval == timedelta(0) - - def test_zero_interval_is_allowed_when_reconnection_is_disabled(self): - """Test that a zero interval is legal when nothing ever reads it.""" - reconnection = TcpReconnectionConfig(enabled=False, interval=timedelta(0)) - - assert reconnection.interval == timedelta(0) + @pytest.mark.parametrize( + "kwargs", + [ + {}, + {"max_retries": 5}, + {"enabled": False}, + ], + ids=["unlimited_retries", "bounded_retries", "reconnection_disabled"], + ) + def test_zero_interval_is_rejected(self, kwargs: dict): + """Test that a zero interval fails whatever the retry policy is. - def test_zero_interval_with_unlimited_retries_is_rejected(self): - """Test that the combination that reconnects in a continuous loop fails.""" + The interval is a delay between attempts, so zero reconnects in a + continuous loop. + """ with pytest.raises(ValueError, match="zero"): - TcpReconnectionConfig(interval=timedelta(0)) + TcpReconnectionConfig(interval=timedelta(0), **kwargs) def test_very_long_interval_round_trips(self): """Test that an interval beyond 68 years survives the i32 boundary.""" @@ -316,32 +317,17 @@ def test_negative_message_expiry_is_rejected(self): [ {"polling_retry_interval": timedelta(0)}, {"init_retries": 3, "init_retry_interval": timedelta(0)}, - {"auto_commit": AutoCommit.Interval(timedelta(0))}, - { - "auto_commit": AutoCommit.IntervalOrWhen( - timedelta(0), AutoCommitWhen.PollingMessages() - ) - }, - { - "auto_commit": AutoCommit.IntervalOrAfter( - timedelta(0), AutoCommitAfter.ConsumingEachMessage() - ) - }, ], ids=[ "polling_retry_interval", "init_retry_interval", - "auto_commit_interval", - "auto_commit_interval_or_when", - "auto_commit_interval_or_after", ], ) def test_zero_consumer_interval_is_rejected(self, interval_kwargs: dict): - """Test that a zero consumer interval fails at the call. + """Test that a zero retry interval fails at the call. - Zero spins the retry loop, floods the server with offset stores, or - panics inside the runtime timer, and none of those name the argument - that caused it. + Zero spins the retry loop or panics inside the runtime timer, and + neither names the argument that caused it. """ client = IggyClient() @@ -353,6 +339,35 @@ def test_zero_consumer_interval_is_rejected(self, interval_kwargs: dict): **interval_kwargs, ) + @pytest.mark.parametrize( + "auto_commit", + [ + AutoCommit.Interval(timedelta(0)), + AutoCommit.IntervalOrWhen(timedelta(0), AutoCommitWhen.PollingMessages()), + AutoCommit.IntervalOrAfter( + timedelta(0), AutoCommitAfter.ConsumingEachMessage() + ), + ], + ids=["interval", "interval_or_when", "interval_or_after"], + ) + def test_zero_auto_commit_interval_is_allowed(self, auto_commit: AutoCommit): + """Test that a zero auto-commit interval passes validation. + + Zero there stores the offsets in a busy loop, which is a supported + choice. Reaching the awaitable is what proves it: building one without + a running loop is the next failure, and a rejected value would have + raised ValueError first. + """ + client = IggyClient() + + with pytest.raises(RuntimeError): + client.consumer_group( + name="group", + stream="stream", + topic="topic", + auto_commit=auto_commit, + ) + def test_zero_poll_interval_is_allowed(self): """Test that a zero poll interval passes validation. From b66b04509b45d8dd62d0535ebc05ab64366780d3 Mon Sep 17 00:00:00 2001 From: Ethan Lin Date: Wed, 19 Aug 2026 23:36:28 +0800 Subject: [PATCH 4/4] refactor(php): carry the non-zero retry interval types The consumer builder takes NonZeroIggyDuration for the retry intervals, so the binding's own zero check hands that type over. --- foreign/php/src/client.rs | 16 +++++----------- 1 file changed, 5 insertions(+), 11 deletions(-) diff --git a/foreign/php/src/client.rs b/foreign/php/src/client.rs index a4413a78c6..ff2b6bbda4 100644 --- a/foreign/php/src/client.rs +++ b/foreign/php/src/client.rs @@ -343,9 +343,8 @@ impl IggyClient { builder = builder.auto_commit(auto_commit.into()); } builder = match poll_interval_micros { - Some(micros) => { - builder.poll_interval(non_zero_duration_micros("poll_interval_micros", micros)?) - } + Some(micros) => builder + .poll_interval(non_zero_duration_micros("poll_interval_micros", micros)?.get()), None => builder.without_poll_interval(), }; if let Some(micros) = polling_retry_interval_micros { @@ -408,12 +407,7 @@ impl IggyClient { } } -fn non_zero_duration_micros(field: &str, micros: u64) -> PhpResult { - if micros == 0 { - return Err(to_php_exception(format!( - "'{field}' must be greater than 0 microseconds" - ))); - } - - Ok(IggyDuration::from(micros)) +fn non_zero_duration_micros(field: &str, micros: u64) -> PhpResult { + NonZeroIggyDuration::try_from(micros) + .map_err(|_| to_php_exception(format!("'{field}' must be greater than 0 microseconds"))) }