This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch feat/durable-protocol-and-sessions in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 8fa1c4a2babe705a053e2380cee51e87275864d0 Author: Hubert Gruszecki <[email protected]> AuthorDate: Sat Oct 3 06:43:56 2026 +0200 fix(sdk): require login after explicit disconnect Explicit QUIC and WebSocket disconnects reused remembered credentials. Clear them like TCP while preserving involuntary transport recovery. Python CI polluted empty storage with logs and expected terminal errors to retry. Move logs beside data and align fixtures and docs with the existing explicit non-admission retry contract. --- core/sdk/src/quic/quic_client.rs | 1 + core/sdk/src/websocket/websocket_client.rs | 1 + foreign/python/README.md | 18 ++++--- foreign/python/apache_iggy.pyi | 13 +++-- foreign/python/src/send_message.rs | 13 +++-- foreign/python/tests/test_producer.py | 76 ++++++++++++++++++------------ 6 files changed, 71 insertions(+), 51 deletions(-) diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs index 1fb8d4059..1c833fac6 100644 --- a/core/sdk/src/quic/quic_client.rs +++ b/core/sdk/src/quic/quic_client.rs @@ -123,6 +123,7 @@ impl Client for QuicClient { } async fn disconnect(&self) -> Result<(), IggyError> { + self.forget_session_credentials().await; QuicClient::disconnect_transport(self).await?; self.reset_vsr_session().await } diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs index 61e3dc687..6afcf6f25 100644 --- a/core/sdk/src/websocket/websocket_client.rs +++ b/core/sdk/src/websocket/websocket_client.rs @@ -117,6 +117,7 @@ impl Client for WebSocketClient { } async fn disconnect(&self) -> Result<(), IggyError> { + self.forget_session_credentials().await; WebSocketClient::disconnect_transport(self).await?; self.reset_vsr_session().await } diff --git a/foreign/python/README.md b/foreign/python/README.md index c727cb5ae..63b2b61f5 100644 --- a/foreign/python/README.md +++ b/foreign/python/README.md @@ -267,12 +267,18 @@ strategy for one call, or `send_to(stream, topic, messages, partitioning)` to send to another existing destination. `send_to()` does not create or initialize that destination. -Direct sends use at-least-once delivery. A request can commit even when its -response is lost, so any retry can write the same batch again. `send_retries` -counts retries after the initial attempt. The first retry runs immediately, and -`send_retry_interval` delays only later retries. Set `send_retries` to `None` or -`0` to disable producer retries. Set `send_retry_interval` to `None` to run all -enabled retries without a delay. A zero interval raises `ValueError`. +Binary transports retain the original request identity after an uncertain +exchange. Persisted sends also provide crash-safe retry receipts; weaker topic +policies can lose data and receipts on a crash. HTTP NoAck confirms dispatch +only. An application resend creates a new request and can duplicate messages. + +Producer retries start a new attempt only after an explicit +`TransientNotAccepted` refusal. Missing destinations and other terminal errors +stop immediately. `send_retries` counts retries after the initial attempt. The +first permitted retry runs immediately, and `send_retry_interval` delays only +later retries. Set `send_retries` to `None` or `0` to disable producer retries. +Set `send_retry_interval` to `None` to run permitted retries without a delay. +A zero interval raises `ValueError`. Transport retries are separate from producer retries. For example, the default `HttpConfig(retries=3)` gives each producer attempt up to four HTTP attempts. diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index c31b6faba..929c1e902 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -2599,9 +2599,8 @@ class SendMessagesConfirmation: r""" Gets the offset assigned to the first message of the batch in this partition. - The offset locates the batch, it does not identify it. Delivery is - at-least-once, so an earlier retry may already have committed these - messages at a lower offset. + The offset locates the batch, it does not identify it. An application + resend creates another request and can duplicate the messages. Confirmation follows VSR quorum commit. A topic with persisted message durability also waits for recoverable stable-storage copies on the quorum. @@ -2620,10 +2619,10 @@ class SendMessagesResponse: The list is empty when the server reports no offsets, so check whether it is empty before indexing into it. - A reported `base_offset` never implies uniqueness, because delivery is - at-least-once and an earlier retry may already have committed the same - messages at a lower offset. Confirmation follows the topic's message - durability policy: quorum commit, plus stable storage for persisted topics. + A reported `base_offset` never implies uniqueness: an application resend + can commit the same messages at another offset. Confirmation follows the + topic's policy: quorum commit, plus stable storage and crash-safe retry + receipts for persisted topics. """ @typing.final diff --git a/foreign/python/src/send_message.rs b/foreign/python/src/send_message.rs index 454fbb2ac..7056a1548 100644 --- a/foreign/python/src/send_message.rs +++ b/foreign/python/src/send_message.rs @@ -148,9 +148,8 @@ impl SendMessagesConfirmation { /// Gets the offset assigned to the first message of the batch in this partition. /// - /// The offset locates the batch, it does not identify it. Delivery is - /// at-least-once, so an earlier retry may already have committed these - /// messages at a lower offset. + /// The offset locates the batch, it does not identify it. An application + /// resend creates another request and can duplicate the messages. /// /// Confirmation follows VSR quorum commit. A topic with persisted message /// durability also waits for recoverable stable-storage copies on the quorum. @@ -181,10 +180,10 @@ impl SendMessagesResponse { /// The list is empty when the server reports no offsets, so check whether /// it is empty before indexing into it. /// - /// A reported `base_offset` never implies uniqueness, because delivery is - /// at-least-once and an earlier retry may already have committed the same - /// messages at a lower offset. Confirmation follows the topic's message - /// durability policy: quorum commit, plus stable storage for persisted topics. + /// A reported `base_offset` never implies uniqueness: an application resend + /// can commit the same messages at another offset. Confirmation follows the + /// topic's policy: quorum commit, plus stable storage and crash-safe retry + /// receipts for persisted topics. #[getter] pub fn confirmations(&self) -> Vec<SendMessagesConfirmation> { self.inner diff --git a/foreign/python/tests/test_producer.py b/foreign/python/tests/test_producer.py index 418b7631e..bdeb31efa 100644 --- a/foreign/python/tests/test_producer.py +++ b/foreign/python/tests/test_producer.py @@ -78,7 +78,7 @@ def spawn_restartable_server( "IGGY_LOGGING_LEVEL": "error", } ) - with (data_path / "server.log").open("ab") as server_log: + with data_path.with_suffix(".log").open("ab") as server_log: return subprocess.Popen( # noqa: S603 [str(SERVER_BINARY), "--with-default-root-credentials"], cwd=REPOSITORY_ROOT, @@ -109,7 +109,7 @@ async def discover_server_address( deadline = time.monotonic() + timeout while time.monotonic() < deadline: if process.poll() is not None: - server_log = (data_path / "server.log").read_text(errors="replace") + server_log = data_path.with_suffix(".log").read_text(errors="replace") raise RuntimeError( f"Restartable Iggy server exited with code {process.returncode}:\n" f"{server_log[-4_000:]}" @@ -1466,51 +1466,60 @@ class TestProducerValidationAndRetries: await producer.shutdown() @pytest.mark.asyncio - async def test_custom_retry_interval_paces_later_retries( + async def test_terminal_error_does_not_wait_for_custom_retry_interval( self, iggy_client: IggyClient, unique_name ): + stream = unique_name() + topic = unique_name() producer = await iggy_client.producer( - unique_name(), - unique_name(), + stream, + topic, send_retries=2, - send_retry_interval=timedelta(milliseconds=150), + send_retry_interval=timedelta(seconds=5), ) try: - started_at = time.monotonic() - with pytest.raises(RuntimeError): - await producer.send_to( - unique_name(), unique_name(), [SendMessage("paced retries")] + with pytest.raises(ProducerSendError) as raised: + await asyncio.wait_for( + producer.send_to( + unique_name(), unique_name(), [SendMessage("terminal failure")] + ), + timeout=1, ) - elapsed = time.monotonic() - started_at - - assert elapsed >= 0.1 + assert raised.value.committed == [] + assert len(raised.value.failed) == 1 + await producer.send(raised.value.failed) + assert await wait_for_payloads( + iggy_client, stream, topic, ["terminal failure"] + ) == ["terminal failure"] finally: await producer.shutdown() @pytest.mark.asyncio - async def test_default_retry_policy_retains_the_rust_interval( + async def test_terminal_error_ignores_default_retry_budget( self, iggy_client: IggyClient, unique_name ): - producer = await iggy_client.producer(unique_name(), unique_name()) + stream = unique_name() + topic = unique_name() + producer = await iggy_client.producer(stream, topic) try: - started_at = time.monotonic() - with pytest.raises(RuntimeError): + with pytest.raises(ProducerSendError) as raised: await asyncio.wait_for( producer.send_to( - unique_name(), unique_name(), [SendMessage("default retries")] + unique_name(), unique_name(), [SendMessage("terminal failure")] ), - timeout=6, + timeout=1, ) - elapsed = time.monotonic() - started_at - - # The first retry is immediate. Three retries with the one-second - # default wait for two later interval ticks. - assert elapsed >= 1.5 + assert raised.value.committed == [] + assert len(raised.value.failed) == 1 + await producer.send(raised.value.failed) + assert await wait_for_payloads( + iggy_client, stream, topic, ["terminal failure"] + ) == ["terminal failure"] finally: await producer.shutdown() @pytest.mark.asyncio - async def test_background_write_retries_until_a_destination_appears( + async def test_background_terminal_error_does_not_wait_for_destination_creation( self, iggy_client: IggyClient, unique_name ): destination_stream = unique_name("retry-stream-", min_bytes=24, max_bytes=24) @@ -1523,7 +1532,7 @@ class TestProducerValidationAndRetries: batch_length=1, ), send_retries=10, - send_retry_interval=timedelta(milliseconds=100), + send_retry_interval=timedelta(seconds=5), ) try: response = await producer.send_to( @@ -1533,14 +1542,19 @@ class TestProducerValidationAndRetries: ) assert response.confirmations == [] - # The first retry is immediate. Create the destination before a - # later interval tick so the worker can recover asynchronously. - await asyncio.sleep(0.03) - await iggy_client.create_stream(destination_stream) - await iggy_client.create_topic(destination_stream, destination_topic, 1) + await asyncio.wait_for(producer.shutdown(), timeout=1) finally: await producer.shutdown() + await iggy_client.create_stream(destination_stream) + await iggy_client.create_topic(destination_stream, destination_topic, 1) + assert ( + await poll_payloads(iggy_client, destination_stream, destination_topic) + == [] + ) + retry = await iggy_client.producer(destination_stream, destination_topic) + async with retry: + await retry.send_one(SendMessage("eventual destination")) assert await wait_for_payloads( iggy_client, destination_stream,
