ethanlin01x commented on code in PR #3991:
URL: https://github.com/apache/iggy/pull/3991#discussion_r3887046467
##########
foreign/python/src/duration.rs:
##########
@@ -40,6 +40,18 @@ pub fn iggy_duration_to_py_delta(
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`).
+pub fn py_delta_to_millis(delta: &Py<PyDelta>) -> PyResult<u64> {
Review Comment:
`as_millis()` truncates, and `0` is a magic value in `configure()`:
```python
QuicConfig(keep_alive_interval=timedelta(microseconds=500)) # -> 0 ->
keep-alive off
QuicConfig(max_idle_timeout=timedelta(microseconds=500)) # -> 0 ->
quinn's 30 s default
```
Please raise `ValueError` when a non-zero duration rounds down to 0 ms.
##########
foreign/python/src/config.rs:
##########
@@ -408,10 +411,413 @@ impl TcpConfig {
}
}
+/// 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: Passes over the known endpoints after the first, or
+ /// `None` for unlimited; `0` still makes that first pass. One pass
+ /// tries the endpoint the client is on, the address it was
+ /// configured with, and every node the roster named, so this
counts
+ /// passes rather than dials. Defaults
+ /// to unlimited, which means a call awaited while the server is
+ /// down never returns: `connect()`, `send_messages()` and
+ /// `poll_messages()` all wait inside the retry loop. Set a finite
+ /// number for request/reply style usage, so a call fails instead.
+ /// interval: Delay between passes. Defaults to 1 second. The first
pass
+ /// runs at once when more than one endpoint is known.
+ /// reestablish_after: Cooldown before redialing the endpoint of the
last
+ /// successful connection, measured from when it was established,
so
+ /// a session that outlived the interval is redialed at once. Owed
to
+ /// that endpoint alone. 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),
+ )
+ }
+}
+
+/// Configuration for the QUIC transport, accepted by `IggyClient.quic(...)`.
+///
+/// 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.
+ /// 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 for no limit. 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` is not a valid `host:port` pair,
if a
+ /// duration is negative, if `heartbeat_interval` is zero, 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 {
+ 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 =
+ u64_param(max_concurrent_bidi_streams,
"max_concurrent_bidi_streams")?;
Review Comment:
`max_concurrent_bidi_streams` and `receive_window` are validated as `u64`,
but they go through `VarInt::try_from` (max `2^62 - 1`).
##########
foreign/python/src/config.rs:
##########
@@ -408,10 +411,413 @@ impl TcpConfig {
}
}
+/// 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: Passes over the known endpoints after the first, or
+ /// `None` for unlimited; `0` still makes that first pass. One pass
+ /// tries the endpoint the client is on, the address it was
+ /// configured with, and every node the roster named, so this
counts
+ /// passes rather than dials. Defaults
+ /// to unlimited, which means a call awaited while the server is
+ /// down never returns: `connect()`, `send_messages()` and
+ /// `poll_messages()` all wait inside the retry loop. Set a finite
+ /// number for request/reply style usage, so a call fails instead.
+ /// interval: Delay between passes. Defaults to 1 second. The first
pass
+ /// runs at once when more than one endpoint is known.
+ /// reestablish_after: Cooldown before redialing the endpoint of the
last
+ /// successful connection, measured from when it was established,
so
+ /// a session that outlived the interval is redialed at once. Owed
to
+ /// that endpoint alone. 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),
+ )
+ }
+}
+
+/// Configuration for the QUIC transport, accepted by `IggyClient.quic(...)`.
+///
+/// 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.
+ /// 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
Review Comment:
The `max_idle_timeout` docs say a zero duration means "no limit". It
actually means "quinn's 30 s default", because `configure()` skips the setter
entirely.
##########
foreign/python/tests/test_quic_config.py:
##########
@@ -0,0 +1,336 @@
+# 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
+
+
[email protected]
+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="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)
+
+
[email protected]
+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)
+
+ 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="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),
+ ],
+ )
+ 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."""
+ with pytest.raises(ValueError, match=field):
+ # pyrefly: ignore # bad-argument-type
+ QuicConfig(**{field: out_of_range})
+
+
[email protected]
+class TestClientConstruction:
+ """Test what `IggyClient.quic(...)` accepts."""
+
+ def test_accepts_a_config(self):
+ """Test that a client can be built from a config object."""
+ assert IggyClient.quic(QuicConfig(server_address="127.0.0.1:8080")) is
not None
+
+ def test_accepts_nothing(self):
+ """Test that the default configuration is used when no argument is
given."""
+ assert IggyClient.quic() is not None
+
+
[email protected]
+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.quic(
+ QuicConfig(
+ server_address=f"{host}:{port}",
+ auto_login=AutoLogin.username_password("iggy", "iggy"),
+ )
+ )
+ await client.connect()
Review Comment:
This and `test_without_auto_login_a_privileged_call_is_unauthenticated` use
the default unlimited reconnection, so a missing QUIC listener hangs until the
CI timeout instead of failing.
`test_wrong_auto_login_credentials_fail` below already passes
`QuicReconnectionConfig(enabled=False)`. Same here?
##########
foreign/python/src/config.rs:
##########
@@ -408,10 +411,413 @@ impl TcpConfig {
}
}
+/// 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: Passes over the known endpoints after the first, or
+ /// `None` for unlimited; `0` still makes that first pass. One pass
+ /// tries the endpoint the client is on, the address it was
+ /// configured with, and every node the roster named, so this
counts
+ /// passes rather than dials. Defaults
+ /// to unlimited, which means a call awaited while the server is
+ /// down never returns: `connect()`, `send_messages()` and
+ /// `poll_messages()` all wait inside the retry loop. Set a finite
+ /// number for request/reply style usage, so a call fails instead.
+ /// interval: Delay between passes. Defaults to 1 second. The first
pass
+ /// runs at once when more than one endpoint is known.
+ /// reestablish_after: Cooldown before redialing the endpoint of the
last
+ /// successful connection, measured from when it was established,
so
+ /// a session that outlived the interval is redialed at once. Owed
to
+ /// that endpoint alone. 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),
+ )
+ }
+}
+
+/// Configuration for the QUIC transport, accepted by `IggyClient.quic(...)`.
+///
+/// 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.
+ /// 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 for no limit. 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` is not a valid `host:port` pair,
if a
+ /// duration is negative, if `heartbeat_interval` is zero, 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 {
+ 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 =
+ u64_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 {
+ inner.initial_mtu = u16_param(initial_mtu, "initial_mtu")?;
Review Comment:
quinn clamps this: `TransportConfig::initial_mtu` does `value.max(1200)`. So
`QuicConfig(initial_mtu=500).initial_mtu` reads back `500` while the connection
runs at 1200. Please reject anything below 1200.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
