💥 Payload limit configuration and validation #1288

cretz · 2026-01-28T20:59:45Z

Technically if the server was comparing sizes of the Payloads message wrapper it is 2-3 bytes larger, but that is probably negligible, though still technically backwards incompatible if this sum is like 3.99999 MB heh.

Probably in the totality of the Payloads object, the size different is nearly negligible, but there are other fields on Payloads that would contribute to the overall size. If contributing non-zero, that means the SDK will pass on potentially too large payloads and the server would reject them, which is the behavior today.

It may be possible to update the visitor in Python to consider allowing the visitor functions to optionally take a Payloads. The visitor functions could have a default implementation that breaks it up into a payload sequence like it does today, but then for this feature it actually provides the override behavior.

there are other fields on Payloads that would contribute to the overall size.

Are there? I am not aware of any, or do you mean in the future? The .ByteSize() invocation is for the whole Payload object, metadata and all

I was looking at https://github.com/temporalio/api-go/blob/c246540cf2eda8f4f8aa9e46ef33d87c48632a27/common/v1/message.pb.go#L81, which is what the server gets the size of e.g. https://github.com/temporalio/temporal/blob/8aeb4bda8dbae17648fb32205cf0b9af62fd07e5/service/frontend/workflow_handler.go#L1236. Maybe proto.Size() knows to ignore the extra fields since they shouldn't contribute to serialized size. So probably not a concern.

Right, that Payloads wrapper is the 2-3 bytes overhead I was talking about, but it only has the one proto field in it (ignore Go fields). There are no extra fields in Payloads, there is only one. We don't support unknown fields, though I guess users could try to hack that.

-Original file line number
+Diff line change
@@ Expand Up / @@ -315,7 +315,6 @@ async def encode_completion( @@
         encode_headers: bool,
     ) -> None:
         """Encode all payloads in the completion."""
-        if data_converter._encode_payload_has_effect:
-            await CommandAwarePayloadVisitor(
-                skip_search_attributes=True, skip_headers=not encode_headers
-            ).visit(_Visitor(data_converter._encode_payload_sequence), completion)
+        await CommandAwarePayloadVisitor(
+            skip_search_attributes=True, skip_headers=not encode_headers
+        ).visit(_Visitor(data_converter._encode_payload_sequence), completion)

-Original file line number
+Diff line change
@@ Expand Up / @@ -6837,7 +6837,7 @@ async def _apply_headers( @@
     ) -> None:
         if source is None:
             return
-        if encode_headers and data_converter._encode_payload_has_effect:
+        if encode_headers:
             for payload in source.values():
                 payload.CopyFrom(await data_converter._encode_payload(payload))
         temporalio.common._apply_headers(source, dest)
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up / @@ -1207,6 +1207,32 @@ def __init__(self) -> None: @@
             super().__init__(encode_common_attributes=True)
+    @dataclass(frozen=True)
+    class PayloadLimitsConfig:
+        """Configuration for when payload sizes exceed limits."""
+        memo_size_warning: int = 2 * 1024
+        """The limit (in bytes) at which a memo size warning is logged."""
+        payload_size_warning: int = 512 * 1024
+        """The limit (in bytes) at which a payload size warning is logged."""
+    class PayloadSizeWarning(RuntimeWarning):
+        """The size of payloads is above the warning limit."""
+    @dataclass(frozen=True)
+    class _ServerPayloadErrorLimits:
+        """Error limits for payloads as described by the Temporal server."""
+        memo_size_error: int
+        """The limit (in bytes) at which a memo size error is raised."""
+        payload_size_error: int
+        """The limit (in bytes) at which a payload size error is raised."""
     @dataclass(frozen=True)
     class DataConverter(WithSerializationContext):
         """Data converter for converting and encoding payloads to/from Python values.
@@ Expand All / @@ -1230,9 +1256,15 @@ 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."""
         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())
@@ Expand Down Expand Up @@
             object.__setattr__(cloned, "failure_converter", failure_converter)
             return cloned
+        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,
@@ Expand Down Expand Up / @@ -1372,30 +1409,38 @@ 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 _encode_payload(
             self, payload: temporalio.api.common.v1.Payload
         ) -> temporalio.api.common.v1.Payload:
             if self.payload_codec:
                 payload = (await self.payload_codec.encode([payload]))[0]
+            self._validate_payload_limits([payload])
             return payload
         async def _encode_payloads(self, payloads: temporalio.api.common.v1.Payloads):
             if self.payload_codec:
                 await self.payload_codec.encode_wrapper(payloads)
+            self._validate_payload_limits(payloads.payloads)
         async def _encode_payload_sequence(
             self, payloads: Sequence[temporalio.api.common.v1.Payload]
         ) -> list[temporalio.api.common.v1.Payload]:
-            if not self.payload_codec:
-                return list(payloads)
-            return await self.payload_codec.encode(payloads)
-        # Temporary shortcircuit detection while the _encode_* methods may no-op if
-        # a payload codec is not configured. Remove once those paths have more to them.
-        @property
-        def _encode_payload_has_effect(self) -> bool:
-            return self.payload_codec is not None
+            encoded_payloads = list(payloads)
+            if self.payload_codec:
+                encoded_payloads = await self.payload_codec.encode(encoded_payloads)
+            self._validate_payload_limits(encoded_payloads)
+            return encoded_payloads
         async def _decode_payload(
             self, payload: temporalio.api.common.v1.Payload
@@ Expand Down Expand Up / @@ -1452,6 +1497,42 @@ async def _apply_to_failure_payloads( @@
             if failure.HasField("cause"):
                 await DataConverter._apply_to_failure_payloads(failure.cause, cb)
+        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 temporalio.exceptions.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,
+                )
     DefaultPayloadConverter.default_encoding_payload_converters = (
         BinaryNullPayloadConverter(),
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up @@
                 and isinstance(exception.cause, CancelledError)
             )
         )
+    class PayloadSizeError(TemporalError):
+        """Error raised when payloads size exceeds payload size limits."""
+        def __init__(self, message: str):
+            """Initialize a payloads size error."""
+            super().__init__(message)
+            self._message = message
+        @property
+        def message(self) -> str:
+            """Message."""
+            return self._message

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

💥 Payload limit configuration and validation #1288

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!

cretz Jan 28, 2026

Uh oh!

jmaeagle99 Jan 29, 2026

Uh oh!

cretz Jan 29, 2026 •

edited

Loading

Uh oh!

jmaeagle99 Jan 29, 2026

Uh oh!

cretz Jan 29, 2026 •

edited

Loading

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

💥 Payload limit configuration and validation #1288

Are you sure you want to change the base?

💥 Payload limit configuration and validation #1288

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!

cretz Jan 28, 2026

Choose a reason for hiding this comment

Uh oh!

jmaeagle99 Jan 29, 2026

Choose a reason for hiding this comment

Uh oh!

cretz Jan 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jmaeagle99 Jan 29, 2026

Choose a reason for hiding this comment

Uh oh!

cretz Jan 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

cretz Jan 29, 2026 •

edited

Loading

cretz Jan 29, 2026 •

edited

Loading