From 4eb9b5c3ebb60e2a63eb589b93b977f4cd0db2bf Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 3 Sep 2026 02:07:13 -0400 Subject: [PATCH 1/2] Retry transient background stream flush failures Signed-off-by: 1fanwang <1fannnw@gmail.com> --- .../contrib/workflow_streams/_client.py | 6 +++- .../workflow_streams/test_workflow_streams.py | 34 +++++++++++++++++++ 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/temporalio/contrib/workflow_streams/_client.py b/temporalio/contrib/workflow_streams/_client.py index 605bf3f03..89e1f62a2 100644 --- a/temporalio/contrib/workflow_streams/_client.py +++ b/temporalio/contrib/workflow_streams/_client.py @@ -463,7 +463,11 @@ async def _run_flusher(self) -> None: except asyncio.TimeoutError: pass self._flush_event.clear() - await self._flush() + try: + await self._flush() + except Exception: + if self._pending is None: + raise @overload def subscribe( diff --git a/tests/contrib/workflow_streams/test_workflow_streams.py b/tests/contrib/workflow_streams/test_workflow_streams.py index e7cedd038..014f91b2d 100644 --- a/tests/contrib/workflow_streams/test_workflow_streams.py +++ b/tests/contrib/workflow_streams/test_workflow_streams.py @@ -1401,6 +1401,40 @@ async def maybe_failing_signal(*args: Any, **kwargs: Any) -> Any: await handle.signal(BasicWorkflowStreamWorkflow.close) +@pytest.mark.asyncio +async def test_background_flusher_retries_failed_signal(client: Client) -> None: + async with new_worker(client, BasicWorkflowStreamWorkflow) as worker: + handle = await client.start_workflow( + BasicWorkflowStreamWorkflow.run, + id=f"workflow-stream-background-flush-retry-{uuid.uuid4()}", + task_queue=worker.task_queue, + ) + + stream = WorkflowStreamClient(handle, batch_interval=timedelta(milliseconds=10)) + real_signal = handle.signal + first_flush_failed = asyncio.Event() + retry_succeeded = asyncio.Event() + + async def fail_first_signal(*args: Any, **kwargs: Any) -> Any: + if not first_flush_failed.is_set(): + first_flush_failed.set() + raise RuntimeError("simulated delivery failure") + result = await real_signal(*args, **kwargs) + retry_succeeded.set() + return result + + with patch.object(handle, "signal", side_effect=fail_first_signal): + async with stream: + stream.topic("events", type=bytes).publish(b"item") + await asyncio.wait_for(first_flush_failed.wait(), timeout=5) + await asyncio.wait_for(retry_succeeded.wait(), timeout=5) + + items = await collect_items(client, handle, None, 0, 1) + assert [item.data for item in items] == [b"item"] + + await handle.signal(BasicWorkflowStreamWorkflow.close) + + @pytest.mark.asyncio async def test_flush_raises_after_max_retry_duration(client: Client) -> None: """When max_retry_duration is exceeded, flush raises TimeoutError and the From 7213ce229b3f4460b2ea2f4c38ab00ba8057f7cf Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 3 Sep 2026 14:36:46 -0400 Subject: [PATCH 2/2] Restrict Workflow Stream background retries Signed-off-by: 1fanwang <1fannnw@gmail.com> --- CHANGELOG.md | 2 + .../contrib/workflow_streams/_client.py | 33 ++++++++- .../workflow_streams/test_workflow_streams.py | 71 ++++++++++++++++++- 3 files changed, 102 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3dc178710..d1b22e95b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,6 +37,8 @@ to include examples, links to docs, or any other relevant information. ### Fixed +- Experimental Workflow Streams background publishing now retries transient + signal delivery failures without delaying payload conversion errors. - `StrandsPlugin` now disables Botocore retries for its default Bedrock model so model request retries are handled exclusively by Temporal. - `temporalio.contrib.openai_agents` now honors the `retry-after-ms` and diff --git a/temporalio/contrib/workflow_streams/_client.py b/temporalio/contrib/workflow_streams/_client.py index 89e1f62a2..cc4010fcd 100644 --- a/temporalio/contrib/workflow_streams/_client.py +++ b/temporalio/contrib/workflow_streams/_client.py @@ -56,6 +56,35 @@ T = TypeVar("T") +# Keep these aligned with SDK Core's client retry policy: +# https://github.com/temporalio/sdk-rust/blob/c8551672bb5c7bf1fb211ef1caf10b2c11483bfc/crates/client/src/retry.rs#L16-L25 +_RETRYABLE_RPC_STATUS_CODES: frozenset[RPCStatusCode] = frozenset( + { + RPCStatusCode.ABORTED, + RPCStatusCode.DATA_LOSS, + RPCStatusCode.INTERNAL, + RPCStatusCode.OUT_OF_RANGE, + RPCStatusCode.RESOURCE_EXHAUSTED, + RPCStatusCode.UNAVAILABLE, + RPCStatusCode.UNKNOWN, + } +) + +# Core forwards oversized-message errors without retrying: +# https://github.com/temporalio/sdk-rust/blob/c8551672bb5c7bf1fb211ef1caf10b2c11483bfc/crates/client/src/retry.rs#L292-L307 +_MESSAGE_TOO_LARGE_ERROR_PREFIXES: tuple[str, ...] = ( + "grpc: received message larger than max", + "grpc: message after decompression larger than max", + "grpc: received message after decompression larger than max", +) + + +def _is_retryable_rpc_error(error: RPCError) -> bool: + return error.status in _RETRYABLE_RPC_STATUS_CODES and not ( + error.status == RPCStatusCode.RESOURCE_EXHAUSTED + and error.message.startswith(_MESSAGE_TOO_LARGE_ERROR_PREFIXES) + ) + class WorkflowStreamClient: """Client for publishing to and subscribing from a workflow stream. @@ -465,8 +494,8 @@ async def _run_flusher(self) -> None: self._flush_event.clear() try: await self._flush() - except Exception: - if self._pending is None: + except RPCError as err: + if not _is_retryable_rpc_error(err): raise @overload diff --git a/tests/contrib/workflow_streams/test_workflow_streams.py b/tests/contrib/workflow_streams/test_workflow_streams.py index 014f91b2d..04f755ea4 100644 --- a/tests/contrib/workflow_streams/test_workflow_streams.py +++ b/tests/contrib/workflow_streams/test_workflow_streams.py @@ -5,6 +5,7 @@ import asyncio import sys import uuid +from collections.abc import Sequence from dataclasses import dataclass from datetime import timedelta from typing import Any, cast @@ -22,6 +23,7 @@ import nexusrpc.handler import pytest +import temporalio.api.common.v1 import temporalio.api.nexus.v1 import temporalio.api.operatorservice.v1 import temporalio.api.workflowservice.v1 @@ -47,9 +49,10 @@ WorkflowTopicHandle, ) from temporalio.contrib.workflow_streams._types import _encode_payload -from temporalio.converter import DataConverter +from temporalio.converter import DataConverter, PayloadCodec from temporalio.exceptions import ApplicationError from temporalio.nexus import WorkflowRunOperationContext, workflow_run_operation +from temporalio.service import RPCError, RPCStatusCode from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker from tests.helpers import assert_eq_eventually, new_worker @@ -67,6 +70,18 @@ def _wire_bytes(data: bytes) -> str: return _encode_payload(payload) +class FailingEncodePayloadCodec(PayloadCodec): + async def encode( + self, payloads: Sequence[temporalio.api.common.v1.Payload] + ) -> list[temporalio.api.common.v1.Payload]: + raise RuntimeError("payload codec encode failed") + + async def decode( + self, payloads: Sequence[temporalio.api.common.v1.Payload] + ) -> list[temporalio.api.common.v1.Payload]: + return list(payloads) + + # --------------------------------------------------------------------------- # Test workflows (must be module-level, not local classes) # --------------------------------------------------------------------------- @@ -1418,7 +1433,11 @@ async def test_background_flusher_retries_failed_signal(client: Client) -> None: async def fail_first_signal(*args: Any, **kwargs: Any) -> Any: if not first_flush_failed.is_set(): first_flush_failed.set() - raise RuntimeError("simulated delivery failure") + raise RPCError( + message="simulated delivery failure", + status=RPCStatusCode.UNAVAILABLE, + raw_grpc_status=b"", + ) result = await real_signal(*args, **kwargs) retry_succeeded.set() return result @@ -1435,6 +1454,54 @@ async def fail_first_signal(*args: Any, **kwargs: Any) -> Any: await handle.signal(BasicWorkflowStreamWorkflow.close) +@pytest.mark.asyncio +async def test_background_flusher_propagates_payload_codec_error( + client: Client, +) -> None: + async with new_worker(client, BasicWorkflowStreamWorkflow) as worker: + handle = await client.start_workflow( + BasicWorkflowStreamWorkflow.run, + id=f"workflow-stream-background-flush-codec-error-{uuid.uuid4()}", + task_queue=worker.task_queue, + ) + config = client.config() + config["data_converter"] = DataConverter( + payload_codec=FailingEncodePayloadCodec() + ) + codec_client = Client(**config) + stream = WorkflowStreamClient.create( + codec_client, + handle.id, + batch_interval=timedelta(milliseconds=10), + ) + stream._flush_task = asyncio.create_task(stream._run_flusher()) + + stream.topic("events", type=bytes).publish(b"item", force_flush=True) + with pytest.raises(RuntimeError, match="payload codec encode failed"): + await asyncio.wait_for(stream._flush_task, timeout=1) + + await handle.signal(BasicWorkflowStreamWorkflow.close) + + +@pytest.mark.asyncio +async def test_background_flusher_propagates_message_too_large( + client: Client, +) -> None: + handle = client.get_workflow_handle("workflow-stream-message-too-large") + stream = WorkflowStreamClient(handle, batch_interval=timedelta(milliseconds=10)) + error = RPCError( + message="grpc: received message larger than max", + status=RPCStatusCode.RESOURCE_EXHAUSTED, + raw_grpc_status=b"", + ) + + with patch.object(handle, "signal", side_effect=error): + flusher = asyncio.create_task(stream._run_flusher()) + stream.topic("events", type=bytes).publish(b"item", force_flush=True) + with pytest.raises(RPCError, match="received message larger than max"): + await asyncio.wait_for(flusher, timeout=1) + + @pytest.mark.asyncio async def test_flush_raises_after_max_retry_duration(client: Client) -> None: """When max_retry_duration is exceeded, flush raises TimeoutError and the