feat(python): add QuicConfig transport configuration (#3991)
Relates to #2835
diff --git a/.github/actions/python-maturin/pre-merge/action.yml b/.github/actions/python-maturin/pre-merge/action.yml
index 234c36e..3cca87f 100644
--- a/.github/actions/python-maturin/pre-merge/action.yml
+++ b/.github/actions/python-maturin/pre-merge/action.yml
@@ -180,6 +180,7 @@
# overwrite the coverage-instrumented .so with a non-instrumented one
IGGY_SERVER_HOST=127.0.0.1 \
IGGY_SERVER_TCP_PORT=8090 \
+ IGGY_SERVER_QUIC_PORT=8080 \
IGGY_SERVER_DOCKER_IMAGE=iggy-server:local \
uv run --no-sync pytest tests/ -v \
--junitxml=../../reports/python-junit.xml \
diff --git a/.github/workflows/coverage-baseline.yml b/.github/workflows/coverage-baseline.yml
index 9daa683..d8f070b 100644
--- a/.github/workflows/coverage-baseline.yml
+++ b/.github/workflows/coverage-baseline.yml
@@ -349,6 +349,7 @@
cd foreign/python
IGGY_SERVER_HOST=127.0.0.1 \
IGGY_SERVER_TCP_PORT=8090 \
+ IGGY_SERVER_QUIC_PORT=8080 \
IGGY_SERVER_DOCKER_IMAGE=iggy-server:local \
uv run --no-sync pytest tests/ -v \
--junitxml=../../reports/python-junit.xml \
diff --git a/core/sdk/src/prelude.rs b/core/sdk/src/prelude.rs
index 72e2503..10cfbc6 100644
--- a/core/sdk/src/prelude.rs
+++ b/core/sdk/src/prelude.rs
@@ -41,6 +41,7 @@
pub use crate::clients::producer_config::{BackgroundConfig, DirectConfig};
pub use crate::clients::producer_sharding::{BalancedSharding, OrderedSharding, Sharding};
pub use crate::consumer_ext::IggyConsumerMessageExt;
+pub use crate::quic::quic_client::QuicClient;
pub use crate::stream_builder::IggyConsumerConfig;
pub use crate::stream_builder::IggyStreamConsumer;
pub use crate::stream_builder::{IggyProducerConfig, IggyStreamProducer};
diff --git a/examples/python/getting-started/consumer.py b/examples/python/getting-started/consumer.py
index 6510841..f3f7886 100755
--- a/examples/python/getting-started/consumer.py
+++ b/examples/python/getting-started/consumer.py
@@ -102,7 +102,22 @@
def build_config(args: ArgNamespace) -> TcpConfig:
- """Build a TCP client configuration with auto-login and reconnection."""
+ """Build the TCP client configuration with auto-login and reconnection."""
+
+ # IggyClient(...) also accepts a QuicConfig for the QUIC transport. To use
+ # it, import QuicConfig and QuicReconnectionConfig above, change the return
+ # annotation to QuicConfig, and replace the return statement with:
+ #
+ # return QuicConfig(
+ # server_address="127.0.0.1:8080",
+ # server_name="localhost",
+ # auto_login=AutoLogin.username_password(args.username, args.password),
+ # reconnection=QuicReconnectionConfig(
+ # enabled=True, interval=timedelta(seconds=1)
+ # ),
+ # )
+ #
+ # main() logs args.tcp_server_address, so change that line too.
return TcpConfig(
server_address=args.tcp_server_address,
diff --git a/examples/python/getting-started/producer.py b/examples/python/getting-started/producer.py
index 80ab7c6..113bee8 100755
--- a/examples/python/getting-started/producer.py
+++ b/examples/python/getting-started/producer.py
@@ -101,7 +101,22 @@
def build_config(args: ArgNamespace) -> TcpConfig:
- """Build a TCP client configuration with auto-login and reconnection."""
+ """Build the TCP client configuration with auto-login and reconnection."""
+
+ # IggyClient(...) also accepts a QuicConfig for the QUIC transport. To use
+ # it, import QuicConfig and QuicReconnectionConfig above, change the return
+ # annotation to QuicConfig, and replace the return statement with:
+ #
+ # return QuicConfig(
+ # server_address="127.0.0.1:8080",
+ # server_name="localhost",
+ # auto_login=AutoLogin.username_password(args.username, args.password),
+ # reconnection=QuicReconnectionConfig(
+ # enabled=True, interval=timedelta(seconds=1)
+ # ),
+ # )
+ #
+ # main() logs args.tcp_server_address, so change that line too.
return TcpConfig(
server_address=args.tcp_server_address,
diff --git a/foreign/python/README.md b/foreign/python/README.md
index e2a4fff..df62cfc 100644
--- a/foreign/python/README.md
+++ b/foreign/python/README.md
@@ -134,7 +134,7 @@
## Client Configuration
-`IggyClient` takes either a server address or a `TcpConfig`:
+`IggyClient` takes a server address, a `TcpConfig`, or a `QuicConfig`:
```python
import asyncio
@@ -168,6 +168,9 @@
asyncio.run(main())
```
+`IggyClient(...)` also accepts a `QuicConfig` for the QUIC transport; see
+`examples/python/getting-started/producer.py` for a config swap example.
+
## Examples
Refer to the [examples/python/](https://github.com/apache/iggy/tree/master/examples/python) directory for usage examples.
diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index ed15ed2..814549f 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -47,6 +47,8 @@
"Partition",
"Permissions",
"PollingStrategy",
+ "QuicConfig",
+ "QuicReconnectionConfig",
"ReceiveMessage",
"SendMessage",
"SendMessagesConfirmation",
@@ -897,23 +899,26 @@
A Python class representing the Iggy client.
It provides asynchronous functionality through the contained runtime.
"""
- def __new__(cls, conn: TcpConfig | builtins.str | None = None) -> IggyClient:
+ def __new__(
+ cls, conn: TcpConfig | QuicConfig | builtins.str | None = None
+ ) -> IggyClient:
r"""
- Constructs a new IggyClient from a TCP server address or a `TcpConfig`.
- This initializes a new runtime for asynchronous operations.
+ Constructs a new IggyClient from a TCP server address, a `TcpConfig`, or a
+ `QuicConfig`. This initializes a new runtime for asynchronous operations.
Future versions might utilize asyncio for more Pythonic async.
Args:
- conn: Either a `host:port` address, or a `TcpConfig` carrying the full
- transport configuration. Defaults to `127.0.0.1:8090` with auto-login
- disabled. A malformed address is reported differently by the two
- forms: the string form raises `RuntimeError` here, while `TcpConfig`
- raises `ValueError` when it is constructed, before it ever reaches
- this call. Neither exception is a subclass of the other.
+ conn: A `host:port` address, a `TcpConfig`, or a `QuicConfig`. Defaults
+ to `127.0.0.1:8090` over TCP with auto-login disabled. A malformed
+ address is reported differently depending on the form: the string
+ form raises `RuntimeError` here, while `TcpConfig`/`QuicConfig`
+ raise `ValueError` when they are constructed, before either ever
+ reaches this call. Neither exception is a subclass of the other.
Raises:
RuntimeError: If the address passed as a string is not a valid
- `host:port` pair.
+ `host:port` pair, or if a `QuicConfig` client cannot bind its
+ local UDP socket (for example the port is already in use).
"""
@classmethod
def from_connection_string(cls, connection_string: builtins.str) -> IggyClient:
@@ -1776,6 +1781,151 @@
...
@typing.final
+class QuicConfig:
+ r"""
+ Configuration for the QUIC transport, accepted by `IggyClient(...)`.
+
+ Every field is keyword-only and optional.
+ """
+ @property
+ def server_address(self) -> builtins.str: ...
+ @property
+ def client_address(self) -> builtins.str: ...
+ @property
+ def server_name(self) -> builtins.str: ...
+ @property
+ def auto_login(self) -> AutoLogin: ...
+ @property
+ def reconnection(self) -> QuicReconnectionConfig: ...
+ @property
+ def heartbeat_interval(self) -> datetime.timedelta: ...
+ @property
+ def response_buffer_size(self) -> builtins.int: ...
+ @property
+ def max_concurrent_bidi_streams(self) -> builtins.int: ...
+ @property
+ def datagram_send_buffer_size(self) -> builtins.int: ...
+ @property
+ def initial_mtu(self) -> builtins.int: ...
+ @property
+ def send_window(self) -> builtins.int: ...
+ @property
+ def receive_window(self) -> builtins.int: ...
+ @property
+ def keep_alive_interval(self) -> datetime.timedelta: ...
+ @property
+ def max_idle_timeout(self) -> datetime.timedelta: ...
+ @property
+ def validate_certificate(self) -> builtins.bool: ...
+ def __new__(
+ cls,
+ *,
+ server_address: builtins.str | None = None,
+ client_address: builtins.str | None = None,
+ server_name: builtins.str | None = None,
+ auto_login: AutoLogin | None = None,
+ reconnection: QuicReconnectionConfig | None = None,
+ heartbeat_interval: datetime.timedelta | None = None,
+ response_buffer_size: builtins.int | None = None,
+ max_concurrent_bidi_streams: builtins.int | None = None,
+ datagram_send_buffer_size: builtins.int | None = None,
+ initial_mtu: builtins.int | None = None,
+ send_window: builtins.int | None = None,
+ receive_window: builtins.int | None = None,
+ keep_alive_interval: datetime.timedelta | None = None,
+ max_idle_timeout: datetime.timedelta | None = None,
+ validate_certificate: builtins.bool | None = None,
+ ) -> QuicConfig:
+ r"""
+ Constructs a QUIC configuration.
+
+ Args:
+ server_address: `host:port` of the Iggy server. Defaults to `127.0.0.1:8080`.
+ client_address: `host:port` to bind the local UDP socket to. Defaults to
+ `127.0.0.1:0`, which binds to any available port. That exact value,
+ passed or defaulted, binds `[::1]:0` instead when `server_address`
+ resolves to IPv6, so the socket in use may not be the address read
+ back here. Any other value binds as given.
+ server_name: Server name used for the QUIC/TLS handshake. Defaults to
+ `localhost`.
+ auto_login: Credentials replayed on every connect. Defaults to `AutoLogin.disabled()`.
+ reconnection: Reconnection policy. Defaults to `QuicReconnectionConfig()`.
+ heartbeat_interval: Interval of heartbeats sent by the client. Defaults to 5 seconds.
+ response_buffer_size: Size of the response buffer in bytes. Defaults to 10 MB.
+ max_concurrent_bidi_streams: Maximum number of concurrent bidirectional
+ streams. Defaults to 10,000.
+ datagram_send_buffer_size: Size of the datagram send buffer in bytes.
+ Defaults to 100,000.
+ initial_mtu: Initial MTU in bytes. Defaults to 1200.
+ send_window: Send window size in bytes. Defaults to 100,000.
+ receive_window: Receive window size in bytes. Defaults to 100,000.
+ keep_alive_interval: Interval between QUIC keep-alive pings, or a zero
+ duration to disable them. Defaults to 5 seconds.
+ max_idle_timeout: How long the connection tolerates silence before it is
+ considered dead, or a zero duration to use quinn's own default (30
+ seconds) instead, since `configure()` skips the setter entirely when
+ zero. Defaults to 10 seconds.
+ validate_certificate: Whether to validate the server certificate. Defaults
+ to disabled, unlike the TCP and WebSocket transports.
+
+ Raises:
+ ValueError: If `server_address` or `client_address` is not a valid
+ `host:port` pair, if a duration is negative, if
+ `heartbeat_interval` is zero, if `keep_alive_interval` or
+ `max_idle_timeout` is not a whole number of milliseconds, if
+ `initial_mtu` is below quinn's minimum of 1200, or if a numeric
+ field is outside the range of its underlying wire type.
+ """
+ def __repr__(self) -> builtins.str: ...
+
+@typing.final
+class QuicReconnectionConfig:
+ r"""
+ How the QUIC client reconnects after the connection to the server is lost.
+ """
+ @property
+ def enabled(self) -> builtins.bool: ...
+ @property
+ def max_retries(self) -> builtins.int | None: ...
+ @property
+ def interval(self) -> datetime.timedelta: ...
+ @property
+ def reestablish_after(self) -> datetime.timedelta: ...
+ def __new__(
+ cls,
+ *,
+ enabled: builtins.bool | None = None,
+ max_retries: builtins.int | None = None,
+ interval: datetime.timedelta | None = None,
+ reestablish_after: datetime.timedelta | None = None,
+ ) -> QuicReconnectionConfig:
+ r"""
+ Constructs a reconnection policy.
+
+ Args:
+ enabled: Whether to reconnect at all. Defaults to enabled.
+ max_retries: Redials of the configured server address after the first
+ attempt, or `None` for unlimited; `0` still makes that first
+ attempt. Unlike the TCP transport, QUIC redials the one address
+ it was configured with rather than walking a cluster roster, so
+ this counts dials. Defaults to unlimited, which means a call
+ awaited while the server is down never returns: `connect()`
+ waits inside the retry loop, as do `send_messages()` and
+ `poll_messages()` once auto-login is configured. Set a finite
+ number for request/reply style usage, so a call fails instead.
+ interval: Delay before each redial. Defaults to 1 second.
+ reestablish_after: Cooldown before redialing after a previously
+ successful connection, measured from when it was established, so
+ a session that outlived the interval is redialed at once.
+ Defaults to 5 seconds.
+
+ 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.
+ """
+ def __repr__(self) -> builtins.str: ...
+
+@typing.final
class ReceiveMessage:
r"""
A Python class representing a received message.
diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index 8252156..f15b6e2 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -28,6 +28,7 @@
use pyo3_stub_gen::define_stub_info_gatherer;
use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods};
use std::collections::BTreeMap;
+use std::fmt::Display;
use std::str::FromStr;
use std::sync::Arc;
@@ -58,6 +59,11 @@
inner: Arc<RustIggyClient>,
}
+/// Keeps the SDK's own message on the `RuntimeError` the Python surface raises.
+fn to_runtime_error<E: Display>(error: E) -> PyErr {
+ PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(error.to_string())
+}
+
/// Resolves the shared `create_topic`/`update_topic` parameters, applying
/// server defaults where the caller left them unset.
fn resolve_topic_params(
@@ -66,8 +72,7 @@
max_topic_size: Option<&MaxTopicSize>,
) -> PyResult<(CompressionAlgorithm, RustIggyExpiry, RustMaxTopicSize)> {
let compression_algorithm = match compression_algorithm {
- Some(algo) => CompressionAlgorithm::from_str(&algo)
- .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?,
+ Some(algo) => CompressionAlgorithm::from_str(&algo).map_err(to_runtime_error)?,
None => CompressionAlgorithm::default(),
};
@@ -87,44 +92,59 @@
#[gen_stub_pymethods]
#[pymethods]
impl IggyClient {
- /// Constructs a new IggyClient from a TCP server address or a `TcpConfig`.
- /// This initializes a new runtime for asynchronous operations.
+ /// Constructs a new IggyClient from a TCP server address, a `TcpConfig`, or a
+ /// `QuicConfig`. This initializes a new runtime for asynchronous operations.
/// Future versions might utilize asyncio for more Pythonic async.
///
/// Args:
- /// conn: Either a `host:port` address, or a `TcpConfig` carrying the full
- /// transport configuration. Defaults to `127.0.0.1:8090` with auto-login
- /// disabled. A malformed address is reported differently by the two
- /// forms: the string form raises `RuntimeError` here, while `TcpConfig`
- /// raises `ValueError` when it is constructed, before it ever reaches
- /// this call. Neither exception is a subclass of the other.
+ /// conn: A `host:port` address, a `TcpConfig`, or a `QuicConfig`. Defaults
+ /// to `127.0.0.1:8090` over TCP with auto-login disabled. A malformed
+ /// address is reported differently depending on the form: the string
+ /// form raises `RuntimeError` here, while `TcpConfig`/`QuicConfig`
+ /// raise `ValueError` when they are constructed, before either ever
+ /// reaches this call. Neither exception is a subclass of the other.
///
/// Raises:
/// RuntimeError: If the address passed as a string is not a valid
- /// `host:port` pair.
+ /// `host:port` pair, or if a `QuicConfig` client cannot bind its
+ /// local UDP socket (for example the port is already in use).
#[new]
#[pyo3(signature = (conn=None))]
fn new(
- #[gen_stub(override_type(type_repr = "TcpConfig | builtins.str | None"))] conn: Option<
- PyClientConfig,
- >,
+ #[gen_stub(override_type(type_repr = "TcpConfig | QuicConfig | builtins.str | None"))]
+ conn: Option<PyClientConfig>,
) -> PyResult<Self> {
- let config = match conn {
- Some(PyClientConfig::Config(config)) => config.client_config(),
- Some(PyClientConfig::ServerAddress(server_address)) => Arc::new(
- TcpClientConfigBuilder::new()
- .with_server_address(server_address)
- .build()
- .map_err(|e| {
- PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string())
- })?,
+ let wrapper = match conn {
+ Some(PyClientConfig::Tcp(config)) => ClientWrapper::Tcp(
+ TcpClient::create(config.client_config()).map_err(to_runtime_error)?,
),
- None => Arc::new(TcpClientConfig::default()),
+ Some(PyClientConfig::ServerAddress(server_address)) => {
+ let config = Arc::new(
+ TcpClientConfigBuilder::new()
+ .with_server_address(server_address)
+ .build()
+ .map_err(to_runtime_error)?,
+ );
+ ClientWrapper::Tcp(TcpClient::create(config).map_err(to_runtime_error)?)
+ }
+ Some(PyClientConfig::Quic(config)) => {
+ // `quinn::Endpoint::client` (invoked eagerly by `QuicClient::create`) looks
+ // up the current Tokio runtime via `Handle::try_current()` and fails with
+ // `CannotCreateEndpoint` if none is active. This method runs synchronously
+ // from Python without one, so enter the runtime pyo3-async-runtimes uses
+ // for our own async methods before building the endpoint.
+ let _guard = pyo3_async_runtimes::tokio::get_runtime().enter();
+ ClientWrapper::Quic(
+ QuicClient::create(config.client_config()).map_err(to_runtime_error)?,
+ )
+ }
+ None => ClientWrapper::Tcp(
+ TcpClient::create(Arc::new(TcpClientConfig::default()))
+ .map_err(to_runtime_error)?,
+ ),
};
- let tcp_client = TcpClient::create(config)
- .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
- Ok(IggyClient {
- inner: Arc::new(RustIggyClient::new(ClientWrapper::Tcp(tcp_client))),
+ Ok(Self {
+ inner: Arc::new(RustIggyClient::new(wrapper)),
})
}
@@ -138,6 +158,11 @@
_cls: &Bound<'_, PyType>,
connection_string: String,
) -> PyResult<Self> {
+ // The QUIC transport builds its endpoint eagerly and needs a Tokio runtime context
+ // to do so (see the `QuicConfig` arm of `new()` above for details); entering it here
+ // is a no-op for the other transports since the protocol isn't known until the
+ // connection string is parsed.
+ let _guard = pyo3_async_runtimes::tokio::get_runtime().enter();
let client = RustIggyClient::from_connection_string(&connection_string)
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
Ok(Self {
diff --git a/foreign/python/src/config.rs b/foreign/python/src/config.rs
index ed6e2a1..cd98ef6 100644
--- a/foreign/python/src/config.rs
+++ b/foreign/python/src/config.rs
@@ -17,6 +17,8 @@
use iggy::prelude::{
AutoLogin as RustAutoLogin, Credentials as RustCredentials,
+ QuicClientConfig as RustQuicClientConfig, QuicClientConfigBuilder,
+ QuicClientReconnectionConfig as RustQuicClientReconnectionConfig,
TcpClientConfig as RustTcpClientConfig, TcpClientConfigBuilder,
TcpClientReconnectionConfig as RustTcpClientReconnectionConfig,
};
@@ -26,10 +28,12 @@
use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods};
use pyo3_stub_gen::impl_stub_type;
use secrecy::SecretString;
+use std::net::SocketAddr;
use std::sync::Arc;
use crate::duration::{
- duration_repr, iggy_duration_to_py_delta, py_delta_to_iggy_duration, reject_zero,
+ duration_repr, iggy_duration_to_py_delta, millis_repr, millis_to_py_delta,
+ py_delta_to_iggy_duration, py_delta_to_millis, reject_zero,
};
/// The credentials replayed by the client every time it (re)connects.
@@ -408,16 +412,469 @@
}
}
+/// How the QUIC client reconnects after the connection to the server is lost.
+#[gen_stub_pyclass]
+#[pyclass(from_py_object)]
+#[derive(Clone)]
+pub struct QuicReconnectionConfig {
+ pub(crate) inner: RustQuicClientReconnectionConfig,
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl QuicReconnectionConfig {
+ /// Constructs a reconnection policy.
+ ///
+ /// Args:
+ /// enabled: Whether to reconnect at all. Defaults to enabled.
+ /// max_retries: Redials of the configured server address after the first
+ /// attempt, or `None` for unlimited; `0` still makes that first
+ /// attempt. Unlike the TCP transport, QUIC redials the one address
+ /// it was configured with rather than walking a cluster roster, so
+ /// this counts dials. Defaults to unlimited, which means a call
+ /// awaited while the server is down never returns: `connect()`
+ /// waits inside the retry loop, as do `send_messages()` and
+ /// `poll_messages()` once auto-login is configured. Set a finite
+ /// number for request/reply style usage, so a call fails instead.
+ /// interval: Delay before each redial. Defaults to 1 second.
+ /// reestablish_after: Cooldown before redialing after a previously
+ /// successful connection, measured from when it was established, so
+ /// a session that outlived the interval is redialed at once.
+ /// Defaults to 5 seconds.
+ ///
+ /// 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.
+ #[new]
+ #[pyo3(signature = (*, enabled=None, max_retries=None, interval=None, reestablish_after=None))]
+ fn new(
+ #[gen_stub(override_type(type_repr = "builtins.bool | None"))] enabled: Option<bool>,
+ #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_retries: Option<i64>,
+ #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
+ interval: Option<Py<PyDelta>>,
+ #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
+ reestablish_after: Option<Py<PyDelta>>,
+ ) -> PyResult<Self> {
+ let defaults = RustQuicClientReconnectionConfig::default();
+ let enabled = enabled.unwrap_or(defaults.enabled);
+ let max_retries = max_retries
+ .map(|max_retries| {
+ u32::try_from(max_retries).map_err(|_| {
+ PyValueError::new_err(format!(
+ "'max_retries' must be between 0 and {}",
+ u32::MAX
+ ))
+ })
+ })
+ .transpose()?;
+ let interval = interval
+ .as_ref()
+ .map(py_delta_to_iggy_duration)
+ .transpose()?
+ .map(|interval| reject_zero(interval, "interval"))
+ .transpose()?
+ .unwrap_or(defaults.interval);
+ Ok(Self {
+ inner: RustQuicClientReconnectionConfig {
+ enabled,
+ max_retries,
+ interval,
+ reestablish_after: reestablish_after
+ .as_ref()
+ .map(py_delta_to_iggy_duration)
+ .transpose()?
+ .unwrap_or(defaults.reestablish_after),
+ },
+ })
+ }
+
+ #[getter]
+ fn enabled(&self) -> bool {
+ self.inner.enabled
+ }
+
+ #[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
+ #[getter]
+ fn max_retries(&self) -> Option<u32> {
+ self.inner.max_retries
+ }
+
+ #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
+ #[getter]
+ fn interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
+ iggy_duration_to_py_delta(py, self.inner.interval.get())
+ }
+
+ #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
+ #[getter]
+ fn reestablish_after<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
+ iggy_duration_to_py_delta(py, self.inner.reestablish_after)
+ }
+
+ fn __repr__(&self) -> String {
+ let max_retries = match self.inner.max_retries {
+ Some(max_retries) => max_retries.to_string(),
+ None => "None".to_owned(),
+ };
+ format!(
+ "QuicReconnectionConfig(enabled={}, max_retries={max_retries}, interval={}, reestablish_after={})",
+ python_bool(self.inner.enabled),
+ duration_repr(self.inner.interval.get()),
+ duration_repr(self.inner.reestablish_after),
+ )
+ }
+}
+
+/// quinn clamps `TransportConfig::initial_mtu` up to this floor rather than
+/// rejecting a smaller value, so `QuicConfig` rejects it instead: otherwise the
+/// getter would read back a value that is not the one actually in effect.
+const QUINN_MIN_INITIAL_MTU: u16 = 1200;
+
+/// Configuration for the QUIC transport, accepted by `IggyClient(...)`.
+///
+/// Every field is keyword-only and optional.
+#[gen_stub_pyclass]
+#[pyclass(from_py_object)]
+#[derive(Clone)]
+pub struct QuicConfig {
+ inner: Arc<RustQuicClientConfig>,
+}
+
+impl QuicConfig {
+ /// The configuration in the shape `QuicClient::create` expects.
+ pub(crate) fn client_config(&self) -> Arc<RustQuicClientConfig> {
+ self.inner.clone()
+ }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl QuicConfig {
+ /// Constructs a QUIC configuration.
+ ///
+ /// Args:
+ /// server_address: `host:port` of the Iggy server. Defaults to `127.0.0.1:8080`.
+ /// client_address: `host:port` to bind the local UDP socket to. Defaults to
+ /// `127.0.0.1:0`, which binds to any available port. That exact value,
+ /// passed or defaulted, binds `[::1]:0` instead when `server_address`
+ /// resolves to IPv6, so the socket in use may not be the address read
+ /// back here. Any other value binds as given.
+ /// server_name: Server name used for the QUIC/TLS handshake. Defaults to
+ /// `localhost`.
+ /// auto_login: Credentials replayed on every connect. Defaults to `AutoLogin.disabled()`.
+ /// reconnection: Reconnection policy. Defaults to `QuicReconnectionConfig()`.
+ /// heartbeat_interval: Interval of heartbeats sent by the client. Defaults to 5 seconds.
+ /// response_buffer_size: Size of the response buffer in bytes. Defaults to 10 MB.
+ /// max_concurrent_bidi_streams: Maximum number of concurrent bidirectional
+ /// streams. Defaults to 10,000.
+ /// datagram_send_buffer_size: Size of the datagram send buffer in bytes.
+ /// Defaults to 100,000.
+ /// initial_mtu: Initial MTU in bytes. Defaults to 1200.
+ /// send_window: Send window size in bytes. Defaults to 100,000.
+ /// receive_window: Receive window size in bytes. Defaults to 100,000.
+ /// keep_alive_interval: Interval between QUIC keep-alive pings, or a zero
+ /// duration to disable them. Defaults to 5 seconds.
+ /// max_idle_timeout: How long the connection tolerates silence before it is
+ /// considered dead, or a zero duration to use quinn's own default (30
+ /// seconds) instead, since `configure()` skips the setter entirely when
+ /// zero. Defaults to 10 seconds.
+ /// validate_certificate: Whether to validate the server certificate. Defaults
+ /// to disabled, unlike the TCP and WebSocket transports.
+ ///
+ /// Raises:
+ /// ValueError: If `server_address` or `client_address` is not a valid
+ /// `host:port` pair, if a duration is negative, if
+ /// `heartbeat_interval` is zero, if `keep_alive_interval` or
+ /// `max_idle_timeout` is not a whole number of milliseconds, if
+ /// `initial_mtu` is below quinn's minimum of 1200, or if a numeric
+ /// field is outside the range of its underlying wire type.
+ #[new]
+ #[pyo3(signature = (
+ *,
+ server_address=None,
+ client_address=None,
+ server_name=None,
+ auto_login=None,
+ reconnection=None,
+ heartbeat_interval=None,
+ response_buffer_size=None,
+ max_concurrent_bidi_streams=None,
+ datagram_send_buffer_size=None,
+ initial_mtu=None,
+ send_window=None,
+ receive_window=None,
+ keep_alive_interval=None,
+ max_idle_timeout=None,
+ validate_certificate=None,
+ ))]
+ #[allow(clippy::too_many_arguments)]
+ fn new(
+ #[gen_stub(override_type(type_repr = "builtins.str | None"))] server_address: Option<
+ String,
+ >,
+ #[gen_stub(override_type(type_repr = "builtins.str | None"))] client_address: Option<
+ String,
+ >,
+ #[gen_stub(override_type(type_repr = "builtins.str | None"))] server_name: Option<String>,
+ #[gen_stub(override_type(type_repr = "AutoLogin | None"))] auto_login: Option<AutoLogin>,
+ #[gen_stub(override_type(type_repr = "QuicReconnectionConfig | None"))]
+ reconnection: Option<QuicReconnectionConfig>,
+ #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
+ heartbeat_interval: Option<Py<PyDelta>>,
+ #[gen_stub(override_type(type_repr = "builtins.int | None"))] response_buffer_size: Option<
+ i64,
+ >,
+ #[gen_stub(override_type(type_repr = "builtins.int | None"))]
+ max_concurrent_bidi_streams: Option<i64>,
+ #[gen_stub(override_type(type_repr = "builtins.int | None"))]
+ datagram_send_buffer_size: Option<i64>,
+ #[gen_stub(override_type(type_repr = "builtins.int | None"))] initial_mtu: Option<i64>,
+ #[gen_stub(override_type(type_repr = "builtins.int | None"))] send_window: Option<i64>,
+ #[gen_stub(override_type(type_repr = "builtins.int | None"))] receive_window: Option<i64>,
+ #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
+ keep_alive_interval: Option<Py<PyDelta>>,
+ #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
+ max_idle_timeout: Option<Py<PyDelta>>,
+ #[gen_stub(override_type(type_repr = "builtins.bool | None"))] validate_certificate: Option<
+ bool,
+ >,
+ ) -> PyResult<Self> {
+ // The builder starts from `QuicClientConfig::default()`, and its `build()`
+ // trims and validates the server address whether or not one was set here.
+ let mut builder = QuicClientConfigBuilder::new();
+ if let Some(server_address) = server_address {
+ builder = builder.with_server_address(server_address);
+ }
+ let mut inner = builder
+ .build()
+ .map_err(|e| PyValueError::new_err(e.to_string()))?;
+ if let Some(client_address) = client_address {
+ // Kept verbatim rather than normalized: `QuicClient::create` compares
+ // this against the literal default to decide whether to bind an IPv6
+ // socket for an IPv6 server, and a rewritten string would not match.
+ client_address.parse::<SocketAddr>().map_err(|e| {
+ PyValueError::new_err(format!("'client_address' is not a valid 'host:port': {e}"))
+ })?;
+ inner.client_address = client_address;
+ }
+ if let Some(server_name) = server_name {
+ inner.server_name = server_name;
+ }
+ if let Some(auto_login) = auto_login {
+ inner.auto_login = auto_login.inner;
+ }
+ if let Some(reconnection) = reconnection {
+ inner.reconnection = reconnection.inner;
+ }
+ if let Some(heartbeat_interval) = heartbeat_interval {
+ inner.heartbeat_interval = reject_zero(
+ py_delta_to_iggy_duration(&heartbeat_interval)?,
+ "heartbeat_interval",
+ )?;
+ }
+ if let Some(response_buffer_size) = response_buffer_size {
+ inner.response_buffer_size = u64_param(response_buffer_size, "response_buffer_size")?;
+ }
+ if let Some(max_concurrent_bidi_streams) = max_concurrent_bidi_streams {
+ inner.max_concurrent_bidi_streams =
+ varint_param(max_concurrent_bidi_streams, "max_concurrent_bidi_streams")?;
+ }
+ if let Some(datagram_send_buffer_size) = datagram_send_buffer_size {
+ inner.datagram_send_buffer_size =
+ u64_param(datagram_send_buffer_size, "datagram_send_buffer_size")?;
+ }
+ if let Some(initial_mtu) = initial_mtu {
+ let initial_mtu = u16_param(initial_mtu, "initial_mtu")?;
+ if initial_mtu < QUINN_MIN_INITIAL_MTU {
+ return Err(PyValueError::new_err(format!(
+ "'initial_mtu' must be at least {QUINN_MIN_INITIAL_MTU}; quinn silently \
+ raises anything smaller to that floor, so the getter would no longer \
+ match the value actually in effect"
+ )));
+ }
+ inner.initial_mtu = initial_mtu;
+ }
+ if let Some(send_window) = send_window {
+ inner.send_window = u64_param(send_window, "send_window")?;
+ }
+ if let Some(receive_window) = receive_window {
+ inner.receive_window = varint_param(receive_window, "receive_window")?;
+ }
+ if let Some(keep_alive_interval) = keep_alive_interval {
+ inner.keep_alive_interval =
+ py_delta_to_millis(&keep_alive_interval, "keep_alive_interval")?;
+ }
+ if let Some(max_idle_timeout) = max_idle_timeout {
+ inner.max_idle_timeout = py_delta_to_millis(&max_idle_timeout, "max_idle_timeout")?;
+ }
+ if let Some(validate_certificate) = validate_certificate {
+ inner.validate_certificate = validate_certificate;
+ }
+
+ Ok(Self {
+ inner: Arc::new(inner),
+ })
+ }
+
+ #[getter]
+ fn server_address(&self) -> String {
+ self.inner.server_address.clone()
+ }
+
+ #[getter]
+ fn client_address(&self) -> String {
+ self.inner.client_address.clone()
+ }
+
+ #[getter]
+ fn server_name(&self) -> String {
+ self.inner.server_name.clone()
+ }
+
+ #[getter]
+ fn auto_login(&self) -> AutoLogin {
+ AutoLogin {
+ inner: self.inner.auto_login.clone(),
+ }
+ }
+
+ #[getter]
+ fn reconnection(&self) -> QuicReconnectionConfig {
+ QuicReconnectionConfig {
+ inner: self.inner.reconnection.clone(),
+ }
+ }
+
+ #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
+ #[getter]
+ fn heartbeat_interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
+ iggy_duration_to_py_delta(py, self.inner.heartbeat_interval.get())
+ }
+
+ #[gen_stub(override_return_type(type_repr = "builtins.int"))]
+ #[getter]
+ fn response_buffer_size(&self) -> u64 {
+ self.inner.response_buffer_size
+ }
+
+ #[gen_stub(override_return_type(type_repr = "builtins.int"))]
+ #[getter]
+ fn max_concurrent_bidi_streams(&self) -> u64 {
+ self.inner.max_concurrent_bidi_streams
+ }
+
+ #[gen_stub(override_return_type(type_repr = "builtins.int"))]
+ #[getter]
+ fn datagram_send_buffer_size(&self) -> u64 {
+ self.inner.datagram_send_buffer_size
+ }
+
+ #[gen_stub(override_return_type(type_repr = "builtins.int"))]
+ #[getter]
+ fn initial_mtu(&self) -> u16 {
+ self.inner.initial_mtu
+ }
+
+ #[gen_stub(override_return_type(type_repr = "builtins.int"))]
+ #[getter]
+ fn send_window(&self) -> u64 {
+ self.inner.send_window
+ }
+
+ #[gen_stub(override_return_type(type_repr = "builtins.int"))]
+ #[getter]
+ fn receive_window(&self) -> u64 {
+ self.inner.receive_window
+ }
+
+ #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
+ #[getter]
+ fn keep_alive_interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
+ millis_to_py_delta(py, self.inner.keep_alive_interval)
+ }
+
+ #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
+ #[getter]
+ fn max_idle_timeout<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
+ millis_to_py_delta(py, self.inner.max_idle_timeout)
+ }
+
+ #[getter]
+ fn validate_certificate(&self) -> bool {
+ self.inner.validate_certificate
+ }
+
+ fn __repr__(&self) -> String {
+ format!(
+ "QuicConfig(server_address={:?}, client_address={:?}, server_name={:?}, auto_login={}, reconnection={}, heartbeat_interval={}, response_buffer_size={}, max_concurrent_bidi_streams={}, datagram_send_buffer_size={}, initial_mtu={}, send_window={}, receive_window={}, keep_alive_interval={}, max_idle_timeout={}, validate_certificate={})",
+ self.inner.server_address,
+ self.inner.client_address,
+ self.inner.server_name,
+ self.auto_login().__repr__(),
+ self.reconnection().__repr__(),
+ duration_repr(self.inner.heartbeat_interval.get()),
+ self.inner.response_buffer_size,
+ self.inner.max_concurrent_bidi_streams,
+ self.inner.datagram_send_buffer_size,
+ self.inner.initial_mtu,
+ self.inner.send_window,
+ self.inner.receive_window,
+ millis_repr(self.inner.keep_alive_interval),
+ millis_repr(self.inner.max_idle_timeout),
+ python_bool(self.inner.validate_certificate),
+ )
+ }
+}
+
fn python_bool(value: bool) -> &'static str {
if value { "True" } else { "False" }
}
-/// What `IggyClient(...)` accepts: a bare `host:port` or a full `TcpConfig`.
+/// Converts a Python int to the unsigned 64-bit integer a QUIC transport
+/// field expects, naming the parameter in the error so a caller can tell
+/// which argument was out of range. The bound in the message is `i64::MAX`
+/// rather than `u64::MAX` because pyo3 extracts the argument as an `i64`
+/// first: anything above that never reaches here, raising `OverflowError`
+/// on the way in. Every one of these fields is a buffer or window size, so
+/// the unreachable half of the range has no practical use.
+fn u64_param(value: i64, parameter: &str) -> PyResult<u64> {
+ u64::try_from(value).map_err(|_| {
+ PyValueError::new_err(format!("'{parameter}' must be between 0 and {}", i64::MAX))
+ })
+}
+
+/// Converts a Python int to the unsigned 16-bit integer `initial_mtu` expects.
+fn u16_param(value: i64, parameter: &str) -> PyResult<u16> {
+ u16::try_from(value).map_err(|_| {
+ PyValueError::new_err(format!("'{parameter}' must be between 0 and {}", u16::MAX))
+ })
+}
+
+/// Converts a Python int to a `u64` that also fits `quinn::VarInt` (max
+/// `2^62 - 1`), which `max_concurrent_bidi_streams` and `receive_window` are
+/// narrowed into when the connection is configured. A `u64` in range for
+/// `u64::MAX` but not `VarInt::MAX` would otherwise only fail there, as an
+/// opaque `RuntimeError` instead of a `ValueError` naming the argument.
+fn varint_param(value: i64, parameter: &str) -> PyResult<u64> {
+ const VARINT_MAX: u64 = (1u64 << 62) - 1;
+ let value = u64_param(value, parameter)?;
+ if value > VARINT_MAX {
+ return Err(PyValueError::new_err(format!(
+ "'{parameter}' must be between 0 and {VARINT_MAX}"
+ )));
+ }
+ Ok(value)
+}
+
+/// What `IggyClient(...)` accepts: a bare `host:port`, a full `TcpConfig`, or a
+/// `QuicConfig` for the QUIC transport.
#[derive(FromPyObject)]
pub enum PyClientConfig {
#[pyo3(transparent)]
- Config(TcpConfig),
+ Tcp(TcpConfig),
+ #[pyo3(transparent)]
+ Quic(QuicConfig),
#[pyo3(transparent, annotation = "str")]
ServerAddress(String),
}
-impl_stub_type!(PyClientConfig = TcpConfig | String);
+impl_stub_type!(PyClientConfig = TcpConfig | QuicConfig | String);
diff --git a/foreign/python/src/duration.rs b/foreign/python/src/duration.rs
index 1d448ad..776df66 100644
--- a/foreign/python/src/duration.rs
+++ b/foreign/python/src/duration.rs
@@ -40,6 +40,34 @@
duration.get_duration().into_pyobject(py)
}
+/// Converts a Python timedelta to milliseconds, for fields the Rust SDK
+/// stores as a raw millisecond count rather than an `IggyDuration` (e.g.
+/// QUIC's `keep_alive_interval`/`max_idle_timeout`). Anything finer than a
+/// millisecond is rejected rather than truncated, so the getter always reads
+/// back the duration that is actually in effect. Zero is accepted for both of
+/// those fields as a magic value (disables the keep-alive, or falls back to
+/// quinn's own default), so a non-zero duration below 1ms is rejected with its
+/// own message rather than collapsing into it.
+pub fn py_delta_to_millis(delta: &Py<PyDelta>, parameter: &str) -> PyResult<u64> {
+ let duration = py_delta_to_iggy_duration(delta)?.get_duration();
+ if !duration.is_zero() && duration.as_millis() == 0 {
+ return Err(PyValueError::new_err(format!(
+ "'{parameter}' is non-zero but rounds down to 0ms; use a duration of at least 1ms, or exactly zero"
+ )));
+ }
+ if duration.subsec_nanos() % 1_000_000 != 0 {
+ return Err(PyValueError::new_err(format!(
+ "'{parameter}' must be a whole number of milliseconds; anything finer is dropped by the QUIC transport, which stores it as a millisecond count"
+ )));
+ }
+ Ok(duration.as_millis() as u64)
+}
+
+/// The inverse of `py_delta_to_millis`.
+pub fn millis_to_py_delta(py: Python<'_>, millis: u64) -> PyResult<Bound<'_, PyDelta>> {
+ Duration::from_millis(millis).into_pyobject(py)
+}
+
/// Renders a duration the way it would be written in Python, so that a `__repr__`
/// built from it can be pasted back into a constructor.
pub fn duration_repr(duration: IggyDuration) -> String {
@@ -53,6 +81,11 @@
}
}
+/// The `duration_repr` equivalent for a raw millisecond count.
+pub fn millis_repr(millis: u64) -> String {
+ duration_repr(IggyDuration::new(Duration::from_millis(millis)))
+}
+
/// 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> {
diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs
index d4397d5..5da3487 100644
--- a/foreign/python/src/lib.rs
+++ b/foreign/python/src/lib.rs
@@ -31,7 +31,7 @@
mod user_headers;
use client::IggyClient;
-use config::{AutoLogin, TcpConfig, TcpReconnectionConfig};
+use config::{AutoLogin, QuicConfig, QuicReconnectionConfig, TcpConfig, TcpReconnectionConfig};
use consumer::{
AutoCommit, AutoCommitAfter, AutoCommitWhen, Consumer, ConsumerGroup, ConsumerGroupDetails,
ConsumerGroupMember, IggyConsumer, ReceiveMessageIterator,
@@ -58,6 +58,8 @@
m.add_class::<AutoLogin>()?;
m.add_class::<TcpConfig>()?;
m.add_class::<TcpReconnectionConfig>()?;
+ m.add_class::<QuicConfig>()?;
+ m.add_class::<QuicReconnectionConfig>()?;
m.add_class::<StreamDetails>()?;
m.add_class::<Stats>()?;
m.add_class::<CacheMetrics>()?;
diff --git a/foreign/python/tests/test_connectivity.py b/foreign/python/tests/test_connectivity.py
index 69516d7..3b42307 100644
--- a/foreign/python/tests/test_connectivity.py
+++ b/foreign/python/tests/test_connectivity.py
@@ -40,6 +40,7 @@
"iggy+http://iggy:iggy@127.0.0.1:3000?heartbeat_interval=5s&retries=3",
"iggy+ws://iggy:iggy@127.0.0.1:8092",
"iggy+ws://iggy:iggy@127.0.0.1:8092?heartbeat_interval=5s&reconnection_retries=3&reconnection_interval=1s&reestablish_after=5s&read_buffer_size=4096&write_buffer_size=4096&max_write_buffer_size=8192&max_message_size=16384&max_frame_size=16384&accept_unmasked_frames=false&tls_domain=localhost&tls_ca_file=unused.pem&tls_validate_certificate=false&tls=false",
+ "iggy+quic://iggy:iggy@127.0.0.1:8080?reconnection_max_retries=0",
],
)
@pytest.mark.asyncio
@@ -77,7 +78,6 @@
"iggy+tcp://iggy:iggy@{host}:{port}?invalid_option=value",
"Invalid connection string",
),
- ("iggy+quic://iggy:iggy@127.0.0.1:8080", "Cannot create endpoint"),
],
)
def test_invalid_connection_string(self, invalid_value: str, expected_error: str):
diff --git a/foreign/python/tests/test_quic_config.py b/foreign/python/tests/test_quic_config.py
new file mode 100644
index 0000000..d0fdc3f
--- /dev/null
+++ b/foreign/python/tests/test_quic_config.py
@@ -0,0 +1,434 @@
+# 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.
+
+"""
+Tests for the QUIC client configuration surface.
+
+`QuicConfig` and `QuicReconnectionConfig` mirror the Rust SDK types the same
+way `TcpConfig`/`TcpReconnectionConfig` do, so most of these assert that a
+value set from Python survives to the getters and that unset fields fall
+back to the Rust defaults. `AutoLogin` is transport-agnostic and already
+covered by `test_client_config.py`.
+"""
+
+import ast
+from collections.abc import Callable
+from datetime import timedelta
+
+import pytest
+
+from apache_iggy import AutoLogin, IggyClient, QuicConfig, QuicReconnectionConfig
+
+from .utils import get_quic_server_config, wait_for_ping
+
+
+@pytest.mark.unit
+class TestQuicReconnectionConfig:
+ """Test the reconnection policy."""
+
+ def test_defaults_match_the_rust_sdk(self):
+ """Test that an unconfigured policy reconnects forever, one second apart."""
+ reconnection = QuicReconnectionConfig()
+
+ assert reconnection.enabled is True
+ assert reconnection.max_retries is None
+ assert reconnection.interval == timedelta(seconds=1)
+ assert reconnection.reestablish_after == timedelta(seconds=5)
+
+ def test_every_field_round_trips(self):
+ """Test that each configured field is readable back unchanged."""
+ reconnection = QuicReconnectionConfig(
+ enabled=False,
+ max_retries=10,
+ interval=timedelta(milliseconds=250),
+ reestablish_after=timedelta(seconds=30),
+ )
+
+ assert reconnection.enabled is False
+ assert reconnection.max_retries == 10
+ assert reconnection.interval == timedelta(milliseconds=250)
+ assert reconnection.reestablish_after == timedelta(seconds=30)
+
+ def test_arguments_are_keyword_only(self):
+ """Test that the adjacent flags cannot be passed positionally."""
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-count
+ QuicReconnectionConfig(True)
+
+ @pytest.mark.parametrize(
+ "construct",
+ [
+ lambda duration: QuicReconnectionConfig(interval=duration),
+ lambda duration: QuicReconnectionConfig(reestablish_after=duration),
+ ],
+ ids=["interval", "reestablish_after"],
+ )
+ @pytest.mark.parametrize(
+ "negative",
+ [timedelta(microseconds=-1), timedelta(seconds=-1), timedelta(days=-1)],
+ )
+ def test_negative_duration_is_rejected(
+ self,
+ construct: Callable[[timedelta], QuicReconnectionConfig],
+ negative: timedelta,
+ ):
+ """Test that a negative duration fails at construction, not at connect."""
+ with pytest.raises(ValueError, match="negative"):
+ construct(negative)
+
+ @pytest.mark.parametrize("out_of_range", [-1, 2**32])
+ def test_out_of_range_max_retries_is_rejected(self, out_of_range: int):
+ """Test that a retry count outside the wire range names the argument.
+
+ The conversion pyo3 does on its own raises OverflowError, which is not a
+ ValueError and so escapes the handler a caller wraps construction in.
+ """
+ with pytest.raises(ValueError, match="max_retries"):
+ QuicReconnectionConfig(max_retries=out_of_range)
+
+ def test_zero_reestablish_after_is_allowed(self):
+ """Test that a zero cooldown is legal and readable back."""
+ reconnection = QuicReconnectionConfig(reestablish_after=timedelta(0))
+
+ assert reconnection.reestablish_after == 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.
+
+ The interval is a delay between passes, so zero reconnects in a
+ continuous loop.
+ """
+ with pytest.raises(ValueError, match=r"interval.*must not be zero"):
+ QuicReconnectionConfig(interval=timedelta(0), **kwargs)
+
+ def test_very_long_interval_round_trips(self):
+ """Test that an interval beyond 68 years survives the i32 boundary."""
+ reconnection = QuicReconnectionConfig(interval=timedelta(days=30_000))
+
+ assert reconnection.interval == timedelta(days=30_000)
+
+ def test_maximum_interval_round_trips(self):
+ """Test that the largest timedelta survives the day conversion."""
+ reconnection = QuicReconnectionConfig(interval=timedelta(days=999_999_999))
+
+ assert reconnection.interval == timedelta(days=999_999_999)
+
+
+@pytest.mark.unit
+class TestQuicConfig:
+ """Test the transport configuration."""
+
+ def test_defaults_match_the_rust_sdk(self):
+ """Test that an unconfigured transport matches the Rust SDK defaults."""
+ config = QuicConfig()
+
+ assert config.server_address == "127.0.0.1:8080"
+ assert config.client_address == "127.0.0.1:0"
+ assert config.server_name == "localhost"
+ assert config.auto_login.enabled is False
+ assert config.reconnection.enabled is True
+ assert config.heartbeat_interval == timedelta(seconds=5)
+ assert config.response_buffer_size == 10_000_000
+ assert config.max_concurrent_bidi_streams == 10_000
+ assert config.datagram_send_buffer_size == 100_000
+ assert config.initial_mtu == 1200
+ assert config.send_window == 100_000
+ assert config.receive_window == 100_000
+ assert config.keep_alive_interval == timedelta(milliseconds=5000)
+ assert config.max_idle_timeout == timedelta(milliseconds=10_000)
+ assert config.validate_certificate is False
+
+ def test_every_field_round_trips(self):
+ """Test that each configured field is readable back unchanged."""
+ config = QuicConfig(
+ server_address="127.0.0.1:8081",
+ client_address="127.0.0.1:9000",
+ server_name="example.com",
+ auto_login=AutoLogin.username_password("iggy", "iggy"),
+ reconnection=QuicReconnectionConfig(max_retries=3),
+ heartbeat_interval=timedelta(seconds=15),
+ response_buffer_size=5_000_000,
+ max_concurrent_bidi_streams=500,
+ datagram_send_buffer_size=50_000,
+ initial_mtu=1400,
+ send_window=200_000,
+ receive_window=200_000,
+ keep_alive_interval=timedelta(seconds=2),
+ max_idle_timeout=timedelta(seconds=20),
+ validate_certificate=True,
+ )
+
+ assert config.server_address == "127.0.0.1:8081"
+ assert config.client_address == "127.0.0.1:9000"
+ assert config.server_name == "example.com"
+ assert config.auto_login.username == "iggy"
+ assert config.reconnection.max_retries == 3
+ assert config.heartbeat_interval == timedelta(seconds=15)
+ assert config.response_buffer_size == 5_000_000
+ assert config.max_concurrent_bidi_streams == 500
+ assert config.datagram_send_buffer_size == 50_000
+ assert config.initial_mtu == 1400
+ assert config.send_window == 200_000
+ assert config.receive_window == 200_000
+ assert config.keep_alive_interval == timedelta(seconds=2)
+ assert config.max_idle_timeout == timedelta(seconds=20)
+ assert config.validate_certificate is True
+
+ def test_arguments_are_keyword_only(self):
+ """Test that the address cannot be passed positionally."""
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-count
+ QuicConfig("127.0.0.1:8080")
+
+ def test_repr_hides_the_password(self):
+ """Test that the password does not leak through repr."""
+ config = QuicConfig(auto_login=AutoLogin.username_password("iggy", "secret"))
+
+ assert "secret" not in repr(config)
+
+ def test_repr_shows_every_field_as_python(self):
+ """Test that repr covers the QUIC-specific fields and parses as Python."""
+ config = QuicConfig(
+ heartbeat_interval=timedelta(seconds=15),
+ keep_alive_interval=timedelta(seconds=2),
+ max_idle_timeout=timedelta(seconds=20),
+ validate_certificate=True,
+ )
+
+ printed = repr(config)
+
+ assert "validate_certificate=True" in printed
+ assert "heartbeat_interval=datetime.timedelta(seconds=15)" in printed
+ assert "keep_alive_interval=datetime.timedelta(seconds=2)" in printed
+ assert "max_idle_timeout=datetime.timedelta(seconds=20)" in printed
+ ast.parse(printed)
+
+ @pytest.mark.parametrize(
+ "invalid_address",
+ ["", "127.0.0.1", "127.0.0.1:not-a-port", "127.0.0.1:70000", "::1:8080"],
+ )
+ def test_invalid_server_address_is_rejected(self, invalid_address: str):
+ """Test that a malformed address fails at construction, not at connect."""
+ with pytest.raises(ValueError):
+ QuicConfig(server_address=invalid_address)
+
+ @pytest.mark.parametrize(
+ "invalid_address",
+ ["", "127.0.0.1", "127.0.0.1:not-a-port", "127.0.0.1:70000", "localhost:0"],
+ )
+ def test_invalid_client_address_is_rejected(self, invalid_address: str):
+ """Test that a malformed bind address fails at construction.
+
+ `QuicClient::create` parses this as a `SocketAddr`, so a hostname is
+ rejected alongside the malformed forms: without the eager check the
+ failure would surface as a `RuntimeError` from `IggyClient(...)`
+ instead, which is not a `ValueError` and so escapes the handler a
+ caller wraps construction in.
+ """
+ with pytest.raises(ValueError, match="client_address"):
+ QuicConfig(client_address=invalid_address)
+
+ def test_negative_heartbeat_interval_is_rejected(self):
+ """Test that a negative heartbeat interval fails at construction."""
+ with pytest.raises(ValueError, match="negative"):
+ QuicConfig(heartbeat_interval=timedelta(seconds=-3))
+
+ def test_zero_heartbeat_interval_is_rejected(self):
+ """Test that a zero heartbeat interval fails at construction.
+
+ Nothing downstream reads zero as "disabled"; it heartbeats in a
+ continuous loop for as long as the client lives.
+ """
+ with pytest.raises(ValueError, match=r"heartbeat_interval.*must not be zero"):
+ QuicConfig(heartbeat_interval=timedelta(0))
+
+ @pytest.mark.parametrize(
+ ("field", "out_of_range"),
+ [
+ ("response_buffer_size", -1),
+ ("max_concurrent_bidi_streams", -1),
+ ("datagram_send_buffer_size", -1),
+ ("send_window", -1),
+ ("receive_window", -1),
+ ("initial_mtu", -1),
+ ("initial_mtu", 2**16),
+ ("max_concurrent_bidi_streams", 2**62),
+ ("receive_window", 2**62),
+ ],
+ )
+ def test_out_of_range_numeric_field_is_rejected(
+ self, field: str, out_of_range: int
+ ):
+ """Test that a numeric field outside its wire type's range names itself.
+
+ `max_concurrent_bidi_streams` and `receive_window` fit `u64`, but
+ quinn narrows them further into a `VarInt` (max `2**62 - 1`), so
+ `2**62` fits the wire type and must still be rejected.
+ """
+ with pytest.raises(ValueError, match=field):
+ # pyrefly: ignore # bad-argument-type
+ QuicConfig(**{field: out_of_range})
+
+ @pytest.mark.parametrize("field", ["keep_alive_interval", "max_idle_timeout"])
+ def test_duration_rounding_down_to_zero_millis_is_rejected(self, field: str):
+ """Test that a non-zero sub-millisecond duration names itself.
+
+ Both fields are raw millisecond counts to the Rust SDK where zero is a
+ magic value (disables the keep-alive, or falls back to quinn's own
+ default), so a duration that rounds down to zero would silently mean
+ something other than what was asked for.
+ """
+ with pytest.raises(ValueError, match=rf"{field}.*rounds down to 0ms"):
+ # pyrefly: ignore # bad-argument-type
+ QuicConfig(**{field: timedelta(microseconds=500)})
+
+ @pytest.mark.parametrize("field", ["keep_alive_interval", "max_idle_timeout"])
+ def test_sub_millisecond_precision_is_rejected(self, field: str):
+ """Test that a duration with a sub-millisecond remainder is refused.
+
+ The Rust SDK stores both fields as a millisecond count, so the
+ remainder would be dropped and the getter would read back a different
+ duration than the one that was passed in.
+ """
+ with pytest.raises(ValueError, match=rf"{field}.*whole number of milliseconds"):
+ # pyrefly: ignore # bad-argument-type
+ QuicConfig(**{field: timedelta(milliseconds=1, microseconds=500)})
+
+ @pytest.mark.parametrize("field", ["keep_alive_interval", "max_idle_timeout"])
+ def test_exact_zero_duration_is_allowed(self, field: str):
+ """Test that an exact zero duration is still legal for these fields."""
+ # pyrefly: ignore # bad-argument-type
+ config = QuicConfig(**{field: timedelta(0)})
+
+ assert getattr(config, field) == timedelta(0)
+
+ @pytest.mark.parametrize("field", ["keep_alive_interval", "max_idle_timeout"])
+ def test_whole_millisecond_duration_is_allowed(self, field: str):
+ """Test that a duration that is not a whole number of seconds round-trips.
+
+ Every other duration these fields accept here is second-aligned, so a
+ check tightened to whole seconds would otherwise pass the suite.
+ """
+ # pyrefly: ignore # bad-argument-type
+ config = QuicConfig(**{field: timedelta(milliseconds=1500)})
+
+ assert getattr(config, field) == timedelta(milliseconds=1500)
+
+ def test_initial_mtu_below_quinns_minimum_is_rejected(self):
+ """Test that an initial_mtu below 1200 fails at construction.
+
+ quinn silently raises anything smaller to that floor instead of
+ rejecting it, so accepting it here would let the getter read back a
+ value that is not the one actually in effect on the connection.
+ """
+ with pytest.raises(ValueError, match="initial_mtu"):
+ QuicConfig(initial_mtu=1199)
+
+ def test_initial_mtu_at_quinns_minimum_is_allowed(self):
+ """Test that exactly 1200, quinn's own floor, is accepted."""
+ config = QuicConfig(initial_mtu=1200)
+
+ assert config.initial_mtu == 1200
+
+
+@pytest.mark.unit
+class TestQuicClientConstruction:
+ """Test that `IggyClient(...)` accepts a `QuicConfig`."""
+
+ def test_accepts_a_config(self):
+ """Test that a client can be built from a config object."""
+ assert IggyClient(QuicConfig(server_address="127.0.0.1:8080")) is not None
+
+ def test_accepts_the_default_config(self):
+ """Test that an explicit default `QuicConfig` is accepted."""
+ assert IggyClient(QuicConfig()) is not None
+
+
+@pytest.mark.integration
+class TestAutoLoginAgainstServer:
+ """Test that configured credentials are actually replayed on connect."""
+
+ @pytest.mark.asyncio
+ async def test_auto_login_authenticates_without_login_user(self, unique_name):
+ """Test that a privileged call succeeds without a manual login_user()."""
+ host, port = get_quic_server_config()
+
+ client = IggyClient(
+ QuicConfig(
+ server_address=f"{host}:{port}",
+ auto_login=AutoLogin.username_password("iggy", "iggy"),
+ # The default reconnection policy retries forever: a missing
+ # listener would stall this test for the full 30s pytest
+ # timeout instead of failing fast.
+ reconnection=QuicReconnectionConfig(enabled=False),
+ )
+ )
+ await client.connect()
+ await wait_for_ping(client)
+
+ stream_name = unique_name()
+ await client.create_stream(stream_name)
+ assert await client.get_stream(stream_name) is not None
+
+ @pytest.mark.asyncio
+ async def test_without_auto_login_a_privileged_call_is_unauthenticated(
+ self, unique_name
+ ):
+ """Test that the same call fails when no credentials are configured."""
+ host, port = get_quic_server_config()
+
+ client = IggyClient(
+ QuicConfig(
+ server_address=f"{host}:{port}",
+ # The default reconnection policy retries forever: a missing
+ # listener would stall this test for the full 30s pytest
+ # timeout instead of failing fast.
+ reconnection=QuicReconnectionConfig(enabled=False),
+ )
+ )
+ await client.connect()
+ await wait_for_ping(client)
+
+ with pytest.raises(RuntimeError):
+ await client.create_stream(unique_name())
+
+ @pytest.mark.asyncio
+ async def test_wrong_auto_login_credentials_fail(self):
+ """Test that bad configured credentials surface as a connect failure."""
+ host, port = get_quic_server_config()
+
+ client = IggyClient(
+ QuicConfig(
+ server_address=f"{host}:{port}",
+ auto_login=AutoLogin.username_password("iggy", "invalid-password"),
+ reconnection=QuicReconnectionConfig(enabled=False),
+ )
+ )
+
+ # A bare RuntimeError would also match "Cannot establish connection",
+ # which is what a missing listener raises on this reconnection policy.
+ with pytest.raises(RuntimeError, match="Invalid credentials"):
+ await client.connect()
diff --git a/foreign/python/tests/utils.py b/foreign/python/tests/utils.py
index b37a538..cdede9f 100644
--- a/foreign/python/tests/utils.py
+++ b/foreign/python/tests/utils.py
@@ -33,15 +33,19 @@
MAX_PASSWORD_BYTES = 100
-def get_server_config() -> tuple[str, int]:
+def get_transport_config(port_env_var: str, default_port: int) -> tuple[str, int]:
"""
- Get server configuration from environment variables or defaults.
+ Get transport-specific server configuration from environment variables or defaults.
+
+ Args:
+ port_env_var: Name of the environment variable holding the port.
+ default_port: Port to use if the environment variable is not set.
Returns:
tuple: (host, port) for the Iggy server
"""
host = os.environ.get("IGGY_SERVER_HOST", "127.0.0.1")
- port = int(os.environ.get("IGGY_SERVER_TCP_PORT", "8090"))
+ port = int(os.environ.get(port_env_var, str(default_port)))
# Convert hostname to IP address for the Rust client
if host not in ("127.0.0.1", "localhost"):
@@ -58,6 +62,26 @@
return host, port
+def get_server_config() -> tuple[str, int]:
+ """
+ Get TCP server configuration from environment variables or defaults.
+
+ Returns:
+ tuple: (host, port) for the Iggy server
+ """
+ return get_transport_config("IGGY_SERVER_TCP_PORT", 8090)
+
+
+def get_quic_server_config() -> tuple[str, int]:
+ """
+ Get QUIC server configuration from environment variables or defaults.
+
+ Returns:
+ tuple: (host, port) for the Iggy server
+ """
+ return get_transport_config("IGGY_SERVER_QUIC_PORT", 8080)
+
+
def wait_for_server(host: str, port: int, timeout: int = 60, interval: int = 2) -> None:
"""
Wait for the server to become available.
