Rename options and move disablement to worker

Author: jmaeagle99

temporalio/converter.py

@@ -1209,43 +1209,23 @@ def __init__(self) -> None:
 
 @dataclass(frozen=True)
 class PayloadLimitsConfig:
-    """Configuration for when uploaded payload sizes exceed the Temporal server's limits."""
+    """Configuration for when payload sizes exceed limits."""
 
-    memo_upload_error_disabled: bool = False
-    """Field indiciating that the memo size checks should be disabled in the SDK.
-
-    A value of False will cause the SDK to fail tasks that attempt to upload memos
-    with a size that is over the Temporal server memo limit. A value of True will
-    disable memo size checks in the SDK, allowing it to attempt to upload memos
-    even if their size is over the Temporal server limit.
-    
-    The default value is False."""
-
-    memo_upload_warning_limit: int = 2 * 1024
+    memo_size_warning: int = 2 * 1024
     """The limit (in bytes) at which a memo size warning is logged."""
 
-    payload_upload_error_disabled: bool = False
-    """Field indiciating that the payload size checks should be disabled in the SDK.
-
-    A value of False will cause the SDK to fail tasks that attempt to upload payloads
-    with a size that is over the Temporal server payloads limit. A value of True will
-    disable payload size checks in the SDK, allowing it to attempt to upload payloads
-    even if their size is over the Temporal server limit.
-    
-    The default value is False."""
-
-    payload_upload_warning_limit: int = 512 * 1024
+    payload_size_warning: int = 512 * 1024
     """The limit (in bytes) at which a payload size warning is logged."""
 
 
 class PayloadSizeWarning(RuntimeWarning):
-    """The size of encoded payloads is above the warning limit."""
+    """The size of payloads is above the warning limit."""
 
 
 @dataclass
 class _PayloadErrorLimits:
-    memo_upload_error_limit: int
-    payload_upload_error_limit: int
+    memo_size_error: int
+    payload_size_error: int
 
 
 @dataclass(frozen=True)
@@ -1277,9 +1257,9 @@ class DataConverter(WithSerializationContext):
     default: ClassVar[DataConverter]
     """Singleton default data converter."""
 
-    _memo_upload_error_limit: int = 0
+    _memo_size_error_limit: int = 0
 
-    _payload_upload_error_limit: int = 0
+    _payload_size_error_limit: int = 0
 
     def __post_init__(self) -> None:  # noqa: D105
         object.__setattr__(self, "payload_converter", self.payload_converter_class())
@@ -1385,12 +1365,8 @@ def with_context(self, context: SerializationContext) -> Self:
     def _with_payload_error_limits(self, options: _PayloadErrorLimits) -> DataConverter:
         return dataclasses.replace(
             self,
-            _memo_upload_error_limit=0
-            if self.payload_limits.memo_upload_error_disabled
-            else options.memo_upload_error_limit,
-            _payload_upload_error_limit=0
-            if self.payload_limits.payload_upload_error_disabled
-            else options.payload_upload_error_limit,
+            _memo_size_error_limit=options.memo_size_error,
+            _payload_size_error_limit=options.payload_size_error,
         )
 
     async def _decode_memo(
@@ -1436,9 +1412,9 @@ async def _encode_memo_existing(
         # Memos have their field payloads validated all together in one unit
         DataConverter._validate_limits(
             payloads,
-            self._memo_upload_error_limit,
+            self._memo_size_error_limit,
             "[TMPRL1103] Attempted to upload memo with size that exceeded the error limit.",
-            self.payload_limits.memo_upload_warning_limit,
+            self.payload_limits.memo_size_warning,
             "[TMPRL1103] Attempted to upload memo with size that exceeded the warning limit.",
         )
 
@@ -1525,9 +1501,9 @@ def _validate_payload_limits(
     ):
         DataConverter._validate_limits(
             payloads,
-            self._payload_upload_error_limit,
+            self._payload_size_error_limit,
             "[TMPRL1103] Attempted to upload payloads with size that exceeded the error limit.",
-            self.payload_limits.payload_upload_warning_limit,
+            self.payload_limits.payload_size_warning,
             "[TMPRL1103] Attempted to upload payloads with size that exceeded the warning limit.",
         )
 

temporalio/worker/_worker.py

@@ -141,6 +141,7 @@ def __init__(
         nexus_task_poller_behavior: PollerBehavior = PollerBehaviorSimpleMaximum(
             maximum=5
         ),
+        disable_payload_error_limit: bool = False,
     ) -> None:
         """Create a worker to process workflows and/or activities.
 
@@ -314,6 +315,14 @@ def __init__(
                 Defaults to a 5-poller maximum.
             nexus_task_poller_behavior: Specify the behavior of Nexus task polling.
                 Defaults to a 5-poller maximum.
+            disable_payload_error_limit: If true, payload and memo error limit checks
+                are disabled in the worker, allowing payloads and memos that are above
+                the server error limit to be submitted to the Temporal server. If false,
+                the worker will validate the size before submitting to the Temporal server,
+                and cause a task failure if the size limit is exceeded. The default is False.
+                See https://docs.temporal.io/troubleshooting/blob-size-limit-error for more
+                details.
+
         """
         config = WorkerConfig(
             client=client,
@@ -357,6 +366,7 @@ def __init__(
             workflow_task_poller_behavior=workflow_task_poller_behavior,
             activity_task_poller_behavior=activity_task_poller_behavior,
             nexus_task_poller_behavior=nexus_task_poller_behavior,
+            disable_payload_error_limit=disable_payload_error_limit,
         )
 
         plugins_from_client = cast(
@@ -719,10 +729,11 @@ async def _run(self):
         namespace_info = await self._bridge_worker.validate()
         payload_error_limits = (
             _PayloadErrorLimits(
-                memo_upload_error_limit=namespace_info.Limits.memo_size_limit_error,
-                payload_upload_error_limit=namespace_info.Limits.blob_size_limit_error,
+                memo_size_error=namespace_info.Limits.memo_size_limit_error,
+                payload_size_error=namespace_info.Limits.blob_size_limit_error,
             )
             if namespace_info.HasField("limits")
+            and not self._config.get("disable_payload_error_limit", False)
             else None
         )
 
@@ -930,6 +941,7 @@ class WorkerConfig(TypedDict, total=False):
     workflow_task_poller_behavior: PollerBehavior
     activity_task_poller_behavior: PollerBehavior
     nexus_task_poller_behavior: PollerBehavior
+    disable_payload_error_limit: bool
 
 
 def _warn_if_activity_executor_max_workers_is_inconsistent(

tests/worker/test_workflow.py

@@ -90,6 +90,7 @@
     PayloadCodec,
     PayloadConverter,
     PayloadLimitsConfig,
+    PayloadSizeWarning,
     _PayloadErrorLimits,
 )
 from temporalio.exceptions import (
@@ -8552,7 +8553,7 @@ async def test_large_payload_workflow_input_warning(client: Client):
     config["data_converter"] = dataclasses.replace(
         temporalio.converter.default(),
         payload_limits=PayloadLimitsConfig(
-            payload_upload_warning_limit=102,
+            payload_size_warning=102,
         ),
     )
     client = Client(**config)
@@ -8575,7 +8576,7 @@ async def test_large_payload_workflow_input_warning(client: Client):
             )
 
         assert len(w) == 1
-        assert issubclass(w[-1].category, UserWarning)
+        assert issubclass(w[-1].category, PayloadSizeWarning)
         assert (
             "[TMPRL1103] Attempted to upload payloads with size that exceeded the warning limit."
             in str(w[-1].message)
@@ -8586,7 +8587,7 @@ async def test_large_payload_workflow_memo_warning(client: Client):
     config = client.config()
     config["data_converter"] = dataclasses.replace(
         temporalio.converter.default(),
-        payload_limits=PayloadLimitsConfig(memo_upload_warning_limit=128),
+        payload_limits=PayloadLimitsConfig(memo_size_warning=128),
     )
     client = Client(**config)
 
@@ -8613,13 +8614,37 @@ async def test_large_payload_workflow_memo_warning(client: Client):
             )
 
         assert len(w) == 1
-        assert issubclass(w[-1].category, UserWarning)
+        assert issubclass(w[-1].category, PayloadSizeWarning)
         assert (
             "[TMPRL1103] Attempted to upload memo with size that exceeded the warning limit."
             in str(w[-1].message)
         )
 
 
+async def test_large_payload_workflow_payload_error_disabled(client: Client):
+    async with new_worker(
+        client,
+        LargePayloadWorkflow,
+        activities=[large_payload_activity],
+        # Referenced server version doesn't report payload limits.
+        # Configure error limits in server when limit reporting supported.
+        disable_payload_error_limit=True,
+    ) as worker:
+        await client.execute_workflow(
+            LargePayloadWorkflow.run,
+            LargePayloadWorkflowInput(
+                activity_input_data_size=0,
+                activity_output_data_size=0,
+                activity_exception_data_size=0,
+                workflow_output_data_size=6 * 1024,
+                data=[],
+            ),
+            id=f"workflow-{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+            execution_timeout=timedelta(seconds=3),
+        )
+
+
 async def test_large_payload_workflow_result_error(client: Client):
     # Create worker runtime with forwarded logger
     worker_logger = logging.getLogger(f"log-{uuid.uuid4()}")
@@ -8642,8 +8667,8 @@ async def test_large_payload_workflow_result_error(client: Client):
         # Remove and configure error limits in server when limit reporting supported.
         data_converter=temporalio.converter.default()._with_payload_error_limits(
             _PayloadErrorLimits(
-                memo_upload_error_limit=0,
-                payload_upload_error_limit=error_limit,
+                memo_size_error=0,
+                payload_size_error=error_limit,
             )
         ),
     )
@@ -8698,7 +8723,7 @@ async def test_large_payload_workflow_result_warning(client: Client):
     config["data_converter"] = dataclasses.replace(
         temporalio.converter.default(),
         payload_limits=PayloadLimitsConfig(
-            payload_upload_warning_limit=1024,
+            payload_size_warning=1024,
         ),
     )
     worker_client = Client(**config)
@@ -8722,7 +8747,7 @@ async def test_large_payload_workflow_result_warning(client: Client):
             )
 
         assert len(w) == 1
-        assert issubclass(w[-1].category, UserWarning)
+        assert issubclass(w[-1].category, PayloadSizeWarning)
         assert (
             "[TMPRL1103] Attempted to upload payloads with size that exceeded the warning limit."
             in str(w[-1].message)
@@ -8751,8 +8776,8 @@ async def test_large_payload_activity_input_error(client: Client):
         # Remove and configure error limits in server when limit reporting supported.
         data_converter=temporalio.converter.default()._with_payload_error_limits(
             _PayloadErrorLimits(
-                memo_upload_error_limit=0,
-                payload_upload_error_limit=error_limit,
+                memo_size_error=0,
+                payload_size_error=error_limit,
             )
         ),
     )
@@ -8806,7 +8831,7 @@ async def test_large_payload_activity_input_warning(client: Client):
     config["data_converter"] = dataclasses.replace(
         temporalio.converter.default(),
         payload_limits=PayloadLimitsConfig(
-            payload_upload_warning_limit=1024,
+            payload_size_warning=1024,
         ),
     )
     worker_client = Client(**config)
@@ -8829,7 +8854,7 @@ async def test_large_payload_activity_input_warning(client: Client):
             )
 
         assert len(w) == 1
-        assert issubclass(w[-1].category, UserWarning)
+        assert issubclass(w[-1].category, PayloadSizeWarning)
         assert (
             "[TMPRL1103] Attempted to upload payloads with size that exceeded the warning limit."
             in str(w[-1].message)
@@ -8858,8 +8883,8 @@ async def test_large_payload_activity_exception_error(client: Client):
         # Remove and configure error limits in server when limit reporting supported.
         data_converter=temporalio.converter.default()._with_payload_error_limits(
             _PayloadErrorLimits(
-                memo_upload_error_limit=0,
-                payload_upload_error_limit=error_limit,
+                memo_size_error=0,
+                payload_size_error=error_limit,
             )
         ),
     )
@@ -8921,8 +8946,8 @@ async def test_large_payload_activity_result_error(client: Client):
         # Remove and configure error limits in server when limit reporting supported.
         data_converter=temporalio.converter.default()._with_payload_error_limits(
             _PayloadErrorLimits(
-                memo_upload_error_limit=0,
-                payload_upload_error_limit=error_limit,
+                memo_size_error=0,
+                payload_size_error=error_limit,
             )
         ),
     )
@@ -8970,7 +8995,7 @@ async def test_large_payload_activity_result_warning(client: Client):
     config["data_converter"] = dataclasses.replace(
         temporalio.converter.default(),
         payload_limits=PayloadLimitsConfig(
-            payload_upload_warning_limit=1024,
+            payload_size_warning=1024,
         ),
     )
     worker_client = Client(**config)
@@ -8993,7 +9018,7 @@ async def test_large_payload_activity_result_warning(client: Client):
             )
 
         assert len(w) == 1
-        assert issubclass(w[-1].category, UserWarning)
+        assert issubclass(w[-1].category, PayloadSizeWarning)
         assert (
             "[TMPRL1103] Attempted to upload payloads with size that exceeded the warning limit."
             in str(w[-1].message)