diff --git a/.github/actions/python-maturin/pre-merge/action.yml b/.github/actions/python-maturin/pre-merge/action.yml index f97462f34d..cedd63cff3 100644 --- a/.github/actions/python-maturin/pre-merge/action.yml +++ b/.github/actions/python-maturin/pre-merge/action.yml @@ -153,6 +153,7 @@ runs: # 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 baa1ae160c..c266d28cd0 100644 --- a/.github/workflows/coverage-baseline.yml +++ b/.github/workflows/coverage-baseline.yml @@ -349,6 +349,7 @@ jobs: 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 81f7d8c16d..22ce243fe9 100644 --- a/core/sdk/src/prelude.rs +++ b/core/sdk/src/prelude.rs @@ -41,6 +41,7 @@ pub use crate::clients::producer_builder::IggyProducerBuilder; 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/README.md b/examples/python/README.md index 7e5da180d4..9d1ec85336 100644 --- a/examples/python/README.md +++ b/examples/python/README.md @@ -94,6 +94,19 @@ python message-headers/typed-headers/producer.py python message-headers/typed-headers/consumer.py ``` +## Transport Protocol Examples + +### QUIC + +Uses the explicit `IggyClient.quic()` constructor. Assumes a server started with defaults, +which enables all four transports (`cargo run --bin iggy-server`, or the `docker run` command +above). + +```bash +uv run quic/producer.py +uv run quic/consumer.py +``` + ## TLS Examples To test with a TLS-enabled server, start the server with TLS configured (see main README), then run: diff --git a/examples/python/quic/consumer.py b/examples/python/quic/consumer.py new file mode 100644 index 0000000000..046483cdff --- /dev/null +++ b/examples/python/quic/consumer.py @@ -0,0 +1,175 @@ +# 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. + +import argparse +import asyncio +import typing +import urllib.parse +from datetime import timedelta + +from apache_iggy import ( + AutoLogin, + Consumer, + IggyClient, + PollingStrategy, + QuicConfig, + QuicReconnectionConfig, + ReceiveMessage, +) +from loguru import logger + +STREAM_NAME = "sample-stream" +TOPIC_NAME = "sample-topic" +STREAM_ID = 0 +TOPIC_ID = 0 +PARTITION_ID = 0 +CONSUMER_NAME = "sample-consumer" +BATCHES_LIMIT = 5 + + +class ArgNamespace(typing.NamedTuple): + server_address: str + server_name: str + username: str + password: str + + +class ValidateUrl(argparse.Action): + def __call__( + self, + parser: argparse.ArgumentParser, + namespace: argparse.Namespace, + values: str, + _option_string: str | None = None, + ): + parsed_url: urllib.parse.ParseResult = urllib.parse.urlparse("//" + values) + if parsed_url.netloc == "" or parsed_url.path != "": + parser.error(f"Invalid server address: {values}") + setattr(namespace, self.dest, values) + + +def parse_args() -> ArgNamespace: + parser = argparse.ArgumentParser() + parser.add_argument( + "--server-address", + help="Iggy QUIC server address (host:port)", + action=ValidateUrl, + default="127.0.0.1:8080", + ) + parser.add_argument( + "--server-name", + default="localhost", + help="Server name used for the QUIC/TLS handshake", + ) + parser.add_argument( + "--username", + default="iggy", + help="Username for authentication", + ) + parser.add_argument( + "--password", + default="iggy", + help="Password for authentication", + ) + args = parser.parse_args() + return ArgNamespace(**vars(args)) + + +def build_config(args: ArgNamespace) -> QuicConfig: + """Build a QUIC client configuration with auto-login and reconnection.""" + + return QuicConfig( + server_address=args.server_address, + server_name=args.server_name, + auto_login=AutoLogin.username_password(args.username, args.password), + reconnection=QuicReconnectionConfig( + enabled=True, + interval=timedelta(seconds=1), + ), + ) + + +async def main(): + args: ArgNamespace = parse_args() + try: + config = build_config(args) + except ValueError as error: + logger.error(f"Invalid client configuration: {error}") + return + logger.info( + f"Connecting to {args.server_address} (server name: {args.server_name})" + ) + + client = IggyClient.quic(config) + try: + logger.info("Connecting to IggyClient...") + # No login_user() call: auto_login replays the credentials on every connect. + await client.connect() + logger.info("Connected.") + await consume_messages(client) + except Exception as error: + logger.exception(f"Exception occurred in main function: {error}") + + +async def consume_messages(client: IggyClient): + interval = 0.5 # 500 milliseconds in seconds for asyncio.sleep + logger.info( + f"Messages will be consumed from stream: {STREAM_NAME}, " + f"topic: {TOPIC_NAME}, partition: {PARTITION_ID} " + f"with interval {interval * 1000} ms." + ) + offset = 0 + messages_per_batch = 10 + n_consumed_batches = 0 + while n_consumed_batches < BATCHES_LIMIT: + try: + logger.debug("Polling for messages...") + polled_messages = await client.poll_messages( + stream=STREAM_NAME, + topic=TOPIC_NAME, + consumer=Consumer.Single(CONSUMER_NAME), + partition_id=PARTITION_ID, + polling_strategy=PollingStrategy.Next(), + count=messages_per_batch, + auto_commit=True, + ) + if not polled_messages: + logger.info("No messages found in current poll") + await asyncio.sleep(interval) + continue + + offset += len(polled_messages) + for message in polled_messages: + handle_message(message) + n_consumed_batches += 1 + await asyncio.sleep(interval) + except Exception as error: + logger.exception(f"Exception occurred while consuming messages: {error}") + break + + logger.info(f"Consumed {n_consumed_batches} batches of messages, exiting.") + + +def handle_message(message: ReceiveMessage): + payload = message.payload().decode("utf-8") + logger.info( + f"Handling message at offset: {message.offset()} with payload: {payload}..." + ) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/python/quic/producer.py b/examples/python/quic/producer.py new file mode 100644 index 0000000000..c3be81ce5c --- /dev/null +++ b/examples/python/quic/producer.py @@ -0,0 +1,190 @@ +# 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. + +import argparse +import asyncio +import typing +import urllib.parse +from datetime import timedelta + +from apache_iggy import ( + AutoLogin, + IggyClient, + QuicConfig, + QuicReconnectionConfig, + StreamDetails, + TopicDetails, +) +from apache_iggy import SendMessage as Message +from loguru import logger + +STREAM_NAME = "sample-stream" +TOPIC_NAME = "sample-topic" +STREAM_ID = 0 +TOPIC_ID = 0 +PARTITION_ID = 0 +BATCHES_LIMIT = 5 + + +class ArgNamespace(typing.NamedTuple): + server_address: str + server_name: str + username: str + password: str + + +class ValidateUrl(argparse.Action): + def __call__( + self, + parser: argparse.ArgumentParser, + namespace: argparse.Namespace, + values: str, + _option_string: str | None = None, + ): + parsed_url: urllib.parse.ParseResult = urllib.parse.urlparse("//" + values) + if parsed_url.netloc == "" or parsed_url.path != "": + parser.error(f"Invalid server address: {values}") + setattr(namespace, self.dest, values) + + +def parse_args() -> ArgNamespace: + parser = argparse.ArgumentParser() + parser.add_argument( + "--server-address", + help="Iggy QUIC server address (host:port)", + action=ValidateUrl, + default="127.0.0.1:8080", + ) + parser.add_argument( + "--server-name", + default="localhost", + help="Server name used for the QUIC/TLS handshake", + ) + parser.add_argument( + "--username", + default="iggy", + help="Username for authentication", + ) + parser.add_argument( + "--password", + default="iggy", + help="Password for authentication", + ) + args = parser.parse_args() + return ArgNamespace(**vars(args)) + + +def build_config(args: ArgNamespace) -> QuicConfig: + """Build a QUIC client configuration with auto-login and reconnection.""" + + return QuicConfig( + server_address=args.server_address, + server_name=args.server_name, + auto_login=AutoLogin.username_password(args.username, args.password), + reconnection=QuicReconnectionConfig( + enabled=True, + interval=timedelta(seconds=1), + ), + ) + + +async def main(): + args: ArgNamespace = parse_args() + try: + config = build_config(args) + except ValueError as error: + logger.error(f"Invalid client configuration: {error}") + return + logger.info( + f"Connecting to {args.server_address} (server name: {args.server_name})" + ) + + client = IggyClient.quic(config) + logger.info("Connecting to IggyClient") + # No login_user() call: auto_login replays the credentials on every connect. + await client.connect() + logger.info("Connected.") + await init_system(client) + await produce_messages(client) + + +async def init_system(client: IggyClient): + logger.info(f"Creating stream with name {STREAM_NAME}...") + stream: StreamDetails | None = await client.get_stream(STREAM_NAME) + if stream is None: + await client.create_stream(name=STREAM_NAME) + logger.info("Stream was created successfully.") + else: + logger.warning(f"Stream {stream.name} already exists with ID {stream.id}") + + logger.info(f"Creating topic {TOPIC_NAME} in stream {STREAM_NAME}") + topic: TopicDetails | None = await client.get_topic(STREAM_NAME, TOPIC_NAME) + if topic is None: + await client.create_topic( + stream=STREAM_NAME, + partitions_count=1, + name=TOPIC_NAME, + ) + logger.info("Topic was created successfully.") + else: + logger.warning(f"Topic {topic.name} already exists with ID {topic.id}") + + +async def produce_messages(client: IggyClient): + interval = 0.5 # 500 milliseconds in seconds for asyncio.sleep + logger.info( + f"Messages will be sent to stream: {STREAM_NAME}, " + f"topic: {TOPIC_NAME}, partition: {PARTITION_ID} " + f"with interval {interval * 1000} ms." + ) + current_id = 0 + messages_per_batch = 10 + n_sent_batches = 0 + while n_sent_batches < BATCHES_LIMIT: + messages = [] + for _ in range(messages_per_batch): + current_id += 1 + payload = f"message-{current_id}" + message = Message(payload) + messages.append(message) + logger.info( + f"Attempting to send batch of {messages_per_batch} messages. " + f"Batch ID: {current_id // messages_per_batch}" + ) + try: + await client.send_messages( + stream=STREAM_NAME, + topic=TOPIC_NAME, + partitioning=PARTITION_ID, + messages=messages, + ) + n_sent_batches += 1 + logger.info( + f"Successfully sent batch of {messages_per_batch} messages. " + f"Batch ID: {current_id // messages_per_batch}" + ) + except Exception as error: + logger.error(f"Exception type: {type(error).__name__}, message: {error}") + logger.exception(error) + break + + await asyncio.sleep(interval) + logger.info(f"Sent {n_sent_batches} batches of messages, exiting.") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/foreign/python/README.md b/foreign/python/README.md index dbbdbe02ca..3c79e56cd6 100644 --- a/foreign/python/README.md +++ b/foreign/python/README.md @@ -166,6 +166,38 @@ async def main(): await client.connect() +asyncio.run(main()) +``` + +`IggyClient.quic(...)` takes a `QuicConfig` the same way, built from `IggyClient.quic()`'s own +config type rather than passed to `IggyClient(...)`: + +```python +import asyncio +from datetime import timedelta + +from apache_iggy import AutoLogin, IggyClient, QuicConfig, QuicReconnectionConfig + + +async def main(): + client = IggyClient.quic( + QuicConfig( + server_address="127.0.0.1:8080", + server_name="localhost", + auto_login=AutoLogin.username_password("iggy", "iggy"), + reconnection=QuicReconnectionConfig( + enabled=True, + max_retries=10, + interval=timedelta(seconds=2), + reestablish_after=timedelta(seconds=30), + ), + heartbeat_interval=timedelta(seconds=5), + # validate_certificate=True, + ) + ) + await client.connect() + + asyncio.run(main()) ``` diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 254b23f41c..11afc99c51 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -45,6 +45,8 @@ __all__ = [ "Partition", "Permissions", "PollingStrategy", + "QuicConfig", + "QuicReconnectionConfig", "ReceiveMessage", "SendMessage", "SendMessagesConfirmation", @@ -867,6 +869,19 @@ class IggyClient: Constructs a new IggyClient from a connection string. Returns an error if the connection string provided is invalid. """ + @classmethod + def quic(cls, config: QuicConfig | None = None) -> IggyClient: + r""" + Constructs a new IggyClient configured for the QUIC transport. + + Args: + config: QUIC transport configuration. Defaults to `QuicConfig()`. + + Raises: + RuntimeError: If the client cannot be constructed, e.g. `client_address` + is not a valid `host:port` pair or the local UDP socket cannot be + bound (for example the port is already in use). + """ def ping(self) -> collections.abc.Awaitable[None]: r""" Sends a ping request to the server to check connectivity. @@ -1704,6 +1719,144 @@ class PollingStrategy: ... +@typing.final +class QuicConfig: + r""" + Configuration for the QUIC transport, accepted by `IggyClient.quic(...)`. + + 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. + 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. + """ + 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: 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. + """ + def __repr__(self) -> builtins.str: ... + @typing.final class ReceiveMessage: r""" diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index f669a87484..416eb39abc 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -31,7 +31,7 @@ use std::collections::BTreeMap; use std::str::FromStr; use std::sync::Arc; -use crate::config::PyClientConfig; +use crate::config::{PyClientConfig, QuicConfig}; use crate::consumer::{ AutoCommit, Consumer as PyConsumer, ConsumerGroup as PyConsumerGroup, ConsumerGroupDetails as PyConsumerGroupDetails, IggyConsumer, @@ -137,6 +137,10 @@ impl IggyClient { _cls: &Bound<'_, PyType>, connection_string: String, ) -> PyResult { + // The QUIC transport builds its endpoint eagerly and needs a Tokio runtime context to do + // so (see `quic()` below 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::(e.to_string()))?; Ok(Self { @@ -144,6 +148,34 @@ impl IggyClient { }) } + /// Constructs a new IggyClient configured for the QUIC transport. + /// + /// Args: + /// config: QUIC transport configuration. Defaults to `QuicConfig()`. + /// + /// Raises: + /// RuntimeError: If the client cannot be constructed, e.g. `client_address` + /// is not a valid `host:port` pair or the local UDP socket cannot be + /// bound (for example the port is already in use). + #[classmethod] + #[pyo3(signature = (config=None))] + fn quic(_cls: &Bound<'_, PyType>, config: Option) -> PyResult { + let config = config + .map(|config| config.client_config()) + .unwrap_or_else(|| Arc::new(QuicClientConfig::default())); + // `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(); + let quic_client = QuicClient::create(config) + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(Self { + inner: Arc::new(RustIggyClient::new(ClientWrapper::Quic(quic_client))), + }) + } + /// Sends a ping request to the server to check connectivity. /// Raises `RuntimeError` if the connection fails. #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))] diff --git a/foreign/python/src/config.rs b/foreign/python/src/config.rs index ed6e2a14f0..dba448b9f2 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, }; @@ -29,7 +31,8 @@ use secrecy::SecretString; 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,10 +411,447 @@ 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, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_retries: Option, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + interval: Option>, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + reestablish_after: Option>, + ) -> PyResult { + 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 { + self.inner.max_retries + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn interval<'a>(&self, py: Python<'a>) -> PyResult> { + iggy_duration_to_py_delta(py, self.inner.interval.get()) + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn reestablish_after<'a>(&self, py: Python<'a>) -> PyResult> { + 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, +} + +impl QuicConfig { + /// The configuration in the shape `QuicClient::create` expects. + pub(crate) fn client_config(&self) -> Arc { + 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 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` 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 non-zero but rounds + /// down to 0ms, 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, + #[gen_stub(override_type(type_repr = "AutoLogin | None"))] auto_login: Option, + #[gen_stub(override_type(type_repr = "QuicReconnectionConfig | None"))] + reconnection: Option, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + heartbeat_interval: Option>, + #[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, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] + datagram_send_buffer_size: Option, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] initial_mtu: Option, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] send_window: Option, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] receive_window: Option, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + keep_alive_interval: Option>, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + max_idle_timeout: Option>, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] validate_certificate: Option< + bool, + >, + ) -> PyResult { + // 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 = + 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> { + 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> { + 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> { + 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" } } +/// 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. +fn u64_param(value: i64, parameter: &str) -> PyResult { + u64::try_from(value).map_err(|_| { + PyValueError::new_err(format!("'{parameter}' must be between 0 and {}", u64::MAX)) + }) +} + +/// Converts a Python int to the unsigned 16-bit integer `initial_mtu` expects. +fn u16_param(value: i64, parameter: &str) -> PyResult { + u16::try_from(value).map_err(|_| { + PyValueError::new_err(format!("'{parameter}' must be between 0 and {}", u16::MAX)) + }) +} + +/// 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; + +/// 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 { + 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` or a full `TcpConfig`. #[derive(FromPyObject)] pub enum PyClientConfig { diff --git a/foreign/python/src/duration.rs b/foreign/python/src/duration.rs index 1d448ad254..979ffe0044 100644 --- a/foreign/python/src/duration.rs +++ b/foreign/python/src/duration.rs @@ -40,6 +40,27 @@ 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`). Zero is a magic value for +/// both of those fields (disables the keep-alive, or falls back to quinn's own +/// default), so a non-zero duration that rounds down to 0ms would silently +/// mean something other than what was asked for. +pub fn py_delta_to_millis(delta: &Py, parameter: &str) -> PyResult { + 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" + ))); + } + Ok(duration.as_millis() as u64) +} + +/// The inverse of `py_delta_to_millis`. +pub fn millis_to_py_delta(py: Python<'_>, millis: u64) -> PyResult> { + 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 +74,11 @@ pub fn duration_repr(duration: IggyDuration) -> String { } } +/// 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 { diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs index 5f2e128264..56ec7f23f8 100644 --- a/foreign/python/src/lib.rs +++ b/foreign/python/src/lib.rs @@ -30,7 +30,7 @@ mod user; 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, @@ -56,6 +56,8 @@ fn apache_iggy(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_class::()?; + m.add_class::()?; + m.add_class::()?; m.add_class::()?; m.add_class::()?; m.add_class::()?; diff --git a/foreign/python/tests/test_connectivity.py b/foreign/python/tests/test_connectivity.py index 69516d74c1..b89751e42a 100644 --- a/foreign/python/tests/test_connectivity.py +++ b/foreign/python/tests/test_connectivity.py @@ -40,6 +40,7 @@ async def test_client_not_none(self, iggy_client: IggyClient): "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", ], ) @pytest.mark.asyncio @@ -77,7 +78,6 @@ async def test_valid_connection_string(self, connection_string: str): "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 0000000000..29927a8337 --- /dev/null +++ b/foreign/python/tests/test_quic_config.py @@ -0,0 +1,392 @@ +# 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="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) + + 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), + ("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=field): + # pyrefly: ignore # bad-argument-type + QuicConfig(**{field: timedelta(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) + + 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 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 + + +@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.quic( + QuicConfig( + server_address=f"{host}:{port}", + auto_login=AutoLogin.username_password("iggy", "iggy"), + # The default reconnection policy retries forever: a missing + # listener would hang this test until the CI timeout instead + # of failing. + 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.quic( + QuicConfig( + server_address=f"{host}:{port}", + # The default reconnection policy retries forever: a missing + # listener would hang this test until the CI timeout instead + # of failing. + 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.quic( + QuicConfig( + server_address=f"{host}:{port}", + auto_login=AutoLogin.username_password("iggy", "invalid-password"), + reconnection=QuicReconnectionConfig(enabled=False), + ) + ) + + with pytest.raises(RuntimeError): + await client.connect() diff --git a/foreign/python/tests/utils.py b/foreign/python/tests/utils.py index b37a53831e..cdede9f32f 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 @@ def get_server_config() -> tuple[str, int]: 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.