Keep Workflow Streams publishing after transient errors

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

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.
@@ -463,7 +492,11 @@ async def _run_flusher(self) -> None:
             except asyncio.TimeoutError:
                 pass
             self._flush_event.clear()
-            await self._flush()
+            try:
+                await self._flush()
+            except RPCError as err:
+                if not _is_retryable_rpc_error(err):
+                    raise
 
     @overload
     def subscribe(

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)
 # ---------------------------------------------------------------------------
@@ -1401,6 +1416,92 @@ 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 RPCError(
+                    message="simulated delivery failure",
+                    status=RPCStatusCode.UNAVAILABLE,
+                    raw_grpc_status=b"",
+                )
+            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_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