💥 refactor: payloads/memo size validation (#1640)

Author: jmaeagle99

CHANGELOG.md

@@ -30,6 +30,12 @@ to include examples, links to docs, or any other relevant information.
 
 ### Breaking Changes
 
+- Payload size limits have moved from `DataConverter` to `Client.connect`. Pass
+  `payload_limits=PayloadLimitsConfig(...)` (now exported from
+  `temporalio.client`) instead of setting `payload_limits` on `DataConverter`.
+  Config fields were renamed to `payloads_warn_size` and `memo_warn_size`, and
+  the deprecated `PayloadSizeWarning` was removed.
+
 ### Fixed
 
 ### Security

temporalio/bridge/client.py

@@ -82,6 +82,8 @@ class ClientConfig:
     http_connect_proxy_config: ClientHttpConnectProxyConfig | None
     dns_load_balancing_config: ClientDnsLoadBalancingConfig | None
     grpc_compression: str
+    payloads_warn_size: int
+    memo_warn_size: int
 
 
 @dataclass

temporalio/bridge/sdk-core

@@ -38,6 +38,8 @@ pub struct ClientConfig {
     http_connect_proxy_config: Option<ClientHttpConnectProxyConfig>,
     dns_load_balancing_config: Option<ClientDnsLoadBalancingConfig>,
     grpc_compression: String,
+    payloads_warn_size: u64,
+    memo_warn_size: u64,
 }
 
 #[derive(FromPyObject)]
@@ -268,6 +270,10 @@ impl ClientConfig {
         .maybe_http_connect_proxy(self.http_connect_proxy_config.map(Into::into))
         .dns_load_balancing(dns_load_balancing)
         .grpc_compression(grpc_compression_from_str(&self.grpc_compression)?)
+        .payload_limits(temporalio_client::PayloadLimitsOptions {
+            payloads_warn_size: self.payloads_warn_size,
+            memo_warn_size: self.memo_warn_size,
+        })
         .headers(ascii_headers)
         .binary_headers(binary_headers)
         .maybe_api_key(self.api_key)

temporalio/bridge/src/client.rs

@@ -63,6 +63,7 @@ pub struct WorkerConfig {
     nexus_task_poller_behavior: PollerBehavior,
     plugins: Vec<String>,
     storage_drivers: HashSet<String>,
+    disable_payload_error_limit: bool,
 }
 
 #[derive(FromPyObject)]
@@ -774,6 +775,7 @@ fn convert_worker_config(
                 .map(|r#type| StorageDriverInfo { r#type })
                 .collect::<HashSet<_>>(),
         )
+        .disable_payload_error_limit(conf.disable_payload_error_limit)
         .build()
         .map_err(|err| PyValueError::new_err(format!("Invalid worker config: {err}")))
 }
@@ -827,11 +829,13 @@ fn convert_tuner_holder(
     }
 
     Ok(temporalio_sdk_core::TunerHolderOptions::builder()
-        .maybe_resource_based_options(first.map(|first| {
-            temporalio_sdk_core::ResourceBasedSlotsOptions::builder()
-                .target_mem_usage(first.target_memory_usage)
-                .target_cpu_usage(first.target_cpu_usage)
-                .build()
+        .maybe_resource_based_config(first.map(|first| {
+            temporalio_sdk_core::ResourceBasedTunerConfig::Options(
+                temporalio_sdk_core::ResourceBasedSlotsOptions::builder()
+                    .target_mem_usage(first.target_memory_usage)
+                    .target_cpu_usage(first.target_cpu_usage)
+                    .build(),
+            )
         }))
         .workflow_slot_options(convert_slot_supplier(
             holder.workflow_slot_supplier,
@@ -899,10 +903,11 @@ fn convert_versioning_strategy(
                     use_worker_versioning: options.use_worker_versioning,
                     default_versioning_behavior: if options.use_worker_versioning {
                         Some(
-                            options
-                                .default_versioning_behavior
-                                .try_into()
-                                .unwrap_or_default(),
+                            temporalio_common::protos::temporal::api::enums::v1::VersioningBehavior::try_from(
+                                options.default_versioning_behavior,
+                            )
+                            .unwrap_or_default()
+                            .into(),
                         )
                     } else {
                         None

temporalio/bridge/src/worker.rs

@@ -58,6 +58,7 @@ class WorkerConfig:
     nexus_task_poller_behavior: PollerBehavior
     plugins: Sequence[str]
     storage_drivers: set[str]
+    disable_payload_error_limit: bool
 
 
 @dataclass
@@ -284,10 +285,8 @@ class _Visitor(VisitorFunctions):
     def __init__(
         self,
         f: Callable[[Sequence[Payload]], Awaitable[list[Payload]]],
-        visit_system_nexus_envelope: Callable[[Payload], Awaitable[None]] | None = None,
     ):
         self._f = f
-        self._visit_system_nexus_envelope = visit_system_nexus_envelope
 
     async def visit_payload(self, payload: Payload) -> None:
         new_payload = (await self._f([payload]))[0]
@@ -303,10 +302,6 @@ async def visit_payloads(self, payloads: PayloadSequence) -> None:
         del payloads[:]
         payloads.extend(new_payloads)
 
-    async def visit_system_nexus_envelope(self, payload: Payload) -> None:
-        if self._visit_system_nexus_envelope is not None:
-            await self._visit_system_nexus_envelope(payload)
-
 
 async def decode_activation(
     activation: temporalio.bridge.proto.workflow_activation.WorkflowActivation,
@@ -348,39 +343,22 @@ async def encode_completion(
     Returns:
         Metrics from any external storage store operations that occurred.
     """
-
-    async def _validate_system_nexus_envelope(payload: Payload) -> None:
-        data_converter._validate_payload_limits([payload])
-
     await CommandAwarePayloadVisitor(
         skip_search_attributes=True,
         skip_headers=not encode_headers,
     ).visit(
-        _Visitor(
-            data_converter._encode_payload_sequence,
-            visit_system_nexus_envelope=_validate_system_nexus_envelope,
-        ),
+        _Visitor(data_converter._encode_payload_sequence),
         completion,
     )
 
-    async def _store_and_validate(
-        payloads: Sequence[Payload],
-    ) -> list[Payload]:
-        stored = await data_converter._external_store_payload_sequence(payloads)
-        data_converter._validate_payload_limits(stored)
-        return stored
-
     metrics = temporalio.converter._extstore.StorageOperationMetrics()
     with metrics.track():
         await CommandAwarePayloadVisitor(
             skip_search_attributes=True,
             skip_headers=not encode_headers,
             concurrency_limit=storage_concurrency_limit,
         ).visit(
-            _Visitor(
-                _store_and_validate,
-                visit_system_nexus_envelope=_validate_system_nexus_envelope,
-            ),
+            _Visitor(data_converter._external_store_payload_sequence),
             completion,
         )
 

temporalio/bridge/worker.py

@@ -19,6 +19,7 @@
     GrpcCompression,
     HttpConnectProxyConfig,
     KeepAliveConfig,
+    PayloadLimitsConfig,
     RetryConfig,
     RPCError,
     RPCStatusCode,
@@ -359,6 +360,7 @@
     "GrpcCompression",
     "HttpConnectProxyConfig",
     "KeepAliveConfig",
+    "PayloadLimitsConfig",
     "RetryConfig",
     "RPCError",
     "RPCStatusCode",

temporalio/client/__init__.py

@@ -35,6 +35,7 @@
     GrpcCompression,
     HttpConnectProxyConfig,
     KeepAliveConfig,
+    PayloadLimitsConfig,
     RetryConfig,
     ServiceClient,
     TLSConfig,
@@ -154,6 +155,7 @@ async def connect(
         http_connect_proxy_config: HttpConnectProxyConfig | None = None,
         dns_load_balancing_config: DnsLoadBalancingConfig | None = None,
         grpc_compression: GrpcCompression = GrpcCompression.GZIP,
+        payload_limits: PayloadLimitsConfig = PayloadLimitsConfig(),
         header_codec_behavior: HeaderCodecBehavior = HeaderCodecBehavior.NO_CODEC,
     ) -> Self:
         """Connect to a Temporal server.
@@ -217,6 +219,8 @@ async def connect(
             grpc_compression: Transport-level gRPC compression for the client
                 connection. Default is gzip. Set to
                 :py:attr:`GrpcCompression.NONE` to disable compression.
+            payload_limits: Warning thresholds for outbound payload/memo sizes. Over-threshold
+                fields are logged but still sent. Set a threshold to 0 to disable it.
             header_codec_behavior: Encoding behavior for headers sent by the client.
         """
         connect_config = temporalio.service.ConnectConfig(
@@ -232,6 +236,7 @@ async def connect(
             http_connect_proxy_config=http_connect_proxy_config,
             dns_load_balancing_config=dns_load_balancing_config,
             grpc_compression=grpc_compression,
+            payload_limits=payload_limits,
         )
 
         def make_lambda(
@@ -3049,6 +3054,7 @@ class ClientConnectConfig(TypedDict, total=False):
     http_connect_proxy_config: HttpConnectProxyConfig | None
     dns_load_balancing_config: DnsLoadBalancingConfig | None
     grpc_compression: GrpcCompression
+    payload_limits: PayloadLimitsConfig
     header_codec_behavior: HeaderCodecBehavior
 
 

temporalio/client/_client.py

@@ -35,10 +35,6 @@
     PayloadConverter,
     value_to_type,
 )
-from temporalio.converter._payload_limits import (
-    PayloadLimitsConfig,
-    PayloadSizeWarning,
-)
 from temporalio.converter._search_attributes import (
     decode_search_attributes,
     decode_typed_search_attributes,
@@ -80,8 +76,6 @@
     "JSONTypeConverterUnhandled",
     "PayloadCodec",
     "PayloadConverter",
-    "PayloadLimitsConfig",
-    "PayloadSizeWarning",
     "SerializationContext",
     "WithSerializationContext",
     "WorkflowSerializationContext",

temporalio/converter/__init__.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import dataclasses
-import warnings
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from logging import getLogger
@@ -30,12 +29,6 @@
 from temporalio.converter._payload_converter import (
     PayloadConverter,
 )
-from temporalio.converter._payload_limits import (
-    PayloadLimitsConfig,
-    PayloadSizeWarning,
-    _PayloadSizeError,
-    _ServerPayloadErrorLimits,
-)
 from temporalio.converter._serialization_context import (
     SerializationContext,
     WithSerializationContext,
@@ -86,9 +79,6 @@ class DataConverter(WithSerializationContext):
     failure_converter: FailureConverter = dataclasses.field(init=False)
     """Failure converter created from the :py:attr:`failure_converter_class`."""
 
-    payload_limits: PayloadLimitsConfig = PayloadLimitsConfig()
-    """Settings for payload size limits."""
-
     external_storage: ExternalStorage | None = None
     """Options for external storage. If None, external storage is disabled.
         
@@ -99,9 +89,6 @@ class DataConverter(WithSerializationContext):
     default: ClassVar[DataConverter]
     """Singleton default data converter."""
 
-    _payload_error_limits: _ServerPayloadErrorLimits | None = None
-    """Server-reported limits for payloads."""
-
     def __post_init__(self) -> None:  # noqa: D105
         object.__setattr__(self, "payload_converter", self.payload_converter_class())
         object.__setattr__(self, "failure_converter", self.failure_converter_class())
@@ -124,7 +111,6 @@ async def encode(
         payloads = self.payload_converter.to_payloads(values)
         payloads = await self._encode_payload_sequence(payloads)
         payloads = await self._external_store_payload_sequence(payloads)
-        self._validate_payload_limits(payloads)
         return payloads
 
     async def decode(
@@ -230,11 +216,6 @@ def _with_contexts(
         """Return an instance with both serialization and store contexts applied."""
         return self.with_context(serialization_ctx)._with_store_context(store_ctx)
 
-    def _with_payload_error_limits(
-        self, limits: _ServerPayloadErrorLimits | None
-    ) -> DataConverter:
-        return dataclasses.replace(self, _payload_error_limits=limits)
-
     async def _decode_memo(
         self,
         source: temporalio.api.common.v1.Memo,
@@ -273,16 +254,6 @@ async def _encode_memo_existing(
             if not isinstance(v, temporalio.api.common.v1.Payload):
                 payload = (await self.encode([v]))[0]
             memo.fields[k].CopyFrom(payload)
-        # Memos have their field payloads validated all together in one unit
-        DataConverter._validate_limits(
-            list(memo.fields.values()),
-            self._payload_error_limits.memo_size_error
-            if self._payload_error_limits
-            else None,
-            "[TMPRL1103] Attempted to upload memo with size that exceeded the error limit.",
-            self.payload_limits.memo_size_warning,
-            "[TMPRL1103] Attempted to upload memo with size that exceeded the warning limit.",
-        )
 
     async def _transform_outbound_payload(
         self, payload: temporalio.api.common.v1.Payload
@@ -291,7 +262,6 @@ async def _transform_outbound_payload(
             payload = (await self.payload_codec.encode([payload]))[0]
         if self.external_storage:
             payload = await self.external_storage._store_payload(payload)
-        self._validate_payload_limits([payload])
         return payload
 
     async def _transform_outbound_payloads(
@@ -301,7 +271,6 @@ async def _transform_outbound_payloads(
             await self.payload_codec.encode_wrapper(payloads)
         if self.external_storage:
             await self.external_storage._store_payloads(payloads)
-        self._validate_payload_limits(payloads.payloads)
 
     async def _transform_inbound_payload(
         self, payload: temporalio.api.common.v1.Payload
@@ -376,42 +345,6 @@ async def _decode_payload_sequence(
     def _decode_payload_has_effect(self) -> bool:
         return self.payload_codec is not None or self.external_storage is not None
 
-    def _validate_payload_limits(
-        self,
-        payloads: Sequence[temporalio.api.common.v1.Payload],
-    ):
-        DataConverter._validate_limits(
-            payloads,
-            self._payload_error_limits.payload_size_error
-            if self._payload_error_limits
-            else None,
-            "[TMPRL1103] Attempted to upload payloads with size that exceeded the error limit.",
-            self.payload_limits.payload_size_warning,
-            "[TMPRL1103] Attempted to upload payloads with size that exceeded the warning limit.",
-        )
-
-    @staticmethod
-    def _validate_limits(
-        payloads: Sequence[temporalio.api.common.v1.Payload],
-        error_limit: int | None,
-        error_message: str,
-        warning_limit: int,
-        warning_message: str,
-    ):
-        total_size = sum(payload.ByteSize() for payload in payloads)
-
-        if error_limit and error_limit > 0 and total_size > error_limit:
-            raise _PayloadSizeError(
-                f"{error_message} Size: {total_size} bytes, Limit: {error_limit} bytes"
-            )
-
-        if warning_limit > 0 and total_size > warning_limit:
-            # TODO: Use a context aware logger to log extra information about workflow/activity/etc
-            warnings.warn(
-                f"{warning_message} Size: {total_size} bytes, Limit: {warning_limit} bytes",
-                PayloadSizeWarning,
-            )
-
 
 def default() -> DataConverter:
     """Default data converter.