diff --git a/temporalio/bridge/client.py b/temporalio/bridge/client.py index c2c5bef6e..cdaf2e178 100644 --- a/temporalio/bridge/client.py +++ b/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 diff --git a/temporalio/bridge/sdk-core b/temporalio/bridge/sdk-core index 467e871aa..56d137b32 160000 --- a/temporalio/bridge/sdk-core +++ b/temporalio/bridge/sdk-core @@ -1 +1 @@ -Subproject commit 467e871aa922ecfeeba8a778b7b9b9de19849acc +Subproject commit 56d137b321156179bccf179ef57e4e33eb987b18 diff --git a/temporalio/bridge/src/client.rs b/temporalio/bridge/src/client.rs index 85dedef94..aaa280a30 100644 --- a/temporalio/bridge/src/client.rs +++ b/temporalio/bridge/src/client.rs @@ -38,6 +38,8 @@ pub struct ClientConfig { http_connect_proxy_config: Option, dns_load_balancing_config: Option, 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) diff --git a/temporalio/bridge/src/worker.rs b/temporalio/bridge/src/worker.rs index e530e89bb..7f38e64dd 100644 --- a/temporalio/bridge/src/worker.rs +++ b/temporalio/bridge/src/worker.rs @@ -63,6 +63,7 @@ pub struct WorkerConfig { nexus_task_poller_behavior: PollerBehavior, plugins: Vec, storage_drivers: HashSet, + disable_payload_error_limit: bool, } #[derive(FromPyObject)] @@ -770,6 +771,7 @@ fn convert_worker_config( .map(|r#type| StorageDriverInfo { r#type }) .collect::>(), ) + .disable_payload_error_limit(conf.disable_payload_error_limit) .build() .map_err(|err| PyValueError::new_err(format!("Invalid worker config: {err}"))) } diff --git a/temporalio/bridge/worker.py b/temporalio/bridge/worker.py index e1e23dd89..6554c508c 100644 --- a/temporalio/bridge/worker.py +++ b/temporalio/bridge/worker.py @@ -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,28 +343,14 @@ 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( @@ -377,10 +358,7 @@ async def _store_and_validate( 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, ) diff --git a/temporalio/client/_client.py b/temporalio/client/_client.py index 5efca9702..8ed4d488d 100644 --- a/temporalio/client/_client.py +++ b/temporalio/client/_client.py @@ -232,6 +232,8 @@ async def connect( http_connect_proxy_config=http_connect_proxy_config, dns_load_balancing_config=dns_load_balancing_config, grpc_compression=grpc_compression, + payloads_warn_size=data_converter.payload_limits.payload_size_warning, + memo_warn_size=data_converter.payload_limits.memo_size_warning, ) def make_lambda( diff --git a/temporalio/converter/_data_converter.py b/temporalio/converter/_data_converter.py index 13b48e695..a27313cc8 100644 --- a/temporalio/converter/_data_converter.py +++ b/temporalio/converter/_data_converter.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 @@ -32,9 +31,6 @@ ) from temporalio.converter._payload_limits import ( PayloadLimitsConfig, - PayloadSizeWarning, - _PayloadSizeError, - _ServerPayloadErrorLimits, ) from temporalio.converter._serialization_context import ( SerializationContext, @@ -99,9 +95,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 +117,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 +222,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 +260,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 +268,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 +277,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 +351,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. diff --git a/temporalio/converter/_failure_converter.py b/temporalio/converter/_failure_converter.py index c76f95c23..848dbc038 100644 --- a/temporalio/converter/_failure_converter.py +++ b/temporalio/converter/_failure_converter.py @@ -17,7 +17,6 @@ import temporalio.api.failure.v1 import temporalio.exceptions from temporalio.converter._payload_converter import PayloadConverter -from temporalio.converter._payload_limits import _PayloadSizeError logger = getLogger("temporalio.converter") @@ -108,9 +107,7 @@ def to_failure( # Convert to failure error failure_error = temporalio.exceptions.ApplicationError( str(exception), - type="PayloadSizeError" - if isinstance(exception, _PayloadSizeError) - else exception.__class__.__name__, + type=exception.__class__.__name__, ) failure_error.__traceback__ = exception.__traceback__ failure_error.__cause__ = exception.__cause__ diff --git a/temporalio/converter/_payload_limits.py b/temporalio/converter/_payload_limits.py index d6eb0b1d2..3ab12dea7 100644 --- a/temporalio/converter/_payload_limits.py +++ b/temporalio/converter/_payload_limits.py @@ -4,44 +4,22 @@ from dataclasses import dataclass -import temporalio.exceptions - @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.""" + """The limit (in bytes) at which a memo size warning is logged. Set to 0 to disable.""" payload_size_warning: int = 512 * 1024 - """The limit (in bytes) at which a payload size warning is logged.""" + """The limit (in bytes) at which a payload size warning is logged. Set to 0 to disable.""" class PayloadSizeWarning(RuntimeWarning): - """The size of payloads is above the warning limit.""" - - -class _PayloadSizeError(temporalio.exceptions.TemporalError): # type:ignore[reportUnusedClass] - """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 - - -@dataclass(frozen=True) -class _ServerPayloadErrorLimits: # type:ignore[reportUnusedClass] - """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.""" + """The size of payloads is above the warning limit. - payload_size_error: int - """The limit (in bytes) at which a payload size error is raised.""" + .. deprecated:: + Payload size warnings are no longer raised through the :mod:`warnings` module. This + symbol is retained for backwards compatibility and is no longer raised. + """ diff --git a/temporalio/runtime.py b/temporalio/runtime.py index 8fab68e9e..94ba9f95d 100644 --- a/temporalio/runtime.py +++ b/temporalio/runtime.py @@ -180,6 +180,7 @@ def formatted(self) -> str: # We intentionally aren't using __str__ or __format__ so they can keep # their original dataclass impls targets = [ + "temporalio_common", "temporalio_sdk_core", "temporalio_client", "temporalio_sdk", diff --git a/temporalio/service.py b/temporalio/service.py index d4cb79720..3c3b4d3a7 100644 --- a/temporalio/service.py +++ b/temporalio/service.py @@ -208,6 +208,8 @@ class ConnectConfig: http_connect_proxy_config: HttpConnectProxyConfig | None = None dns_load_balancing_config: DnsLoadBalancingConfig | None = None grpc_compression: GrpcCompression = GrpcCompression.GZIP + payloads_warn_size: int = 512 * 1024 + memo_warn_size: int = 2 * 1024 def __post_init__(self) -> None: """Set extra defaults on unset properties.""" @@ -271,6 +273,8 @@ def _to_bridge_config(self) -> temporalio.bridge.client.ClientConfig: else None ), grpc_compression=self.grpc_compression._to_bridge_config(), + payloads_warn_size=self.payloads_warn_size, + memo_warn_size=self.memo_warn_size, ) diff --git a/temporalio/worker/_activity.py b/temporalio/worker/_activity.py index 088ed0380..3ad572d4b 100644 --- a/temporalio/worker/_activity.py +++ b/temporalio/worker/_activity.py @@ -32,7 +32,6 @@ import temporalio.client import temporalio.common import temporalio.converter -import temporalio.converter._payload_limits import temporalio.exceptions from temporalio.converter import ( StorageDriverActivityInfo, @@ -132,15 +131,8 @@ def __init__( else: self._dynamic_activity = defn - async def run( - self, - payload_error_limits: temporalio.converter._payload_limits._ServerPayloadErrorLimits - | None, - ) -> None: + async def run(self) -> None: """Continually poll for activity tasks and dispatch to handlers.""" - self._data_converter = self._data_converter._with_payload_error_limits( - payload_error_limits - ) async def raise_from_exception_queue() -> NoReturn: raise await self._fail_worker_exception_queue.get() @@ -363,136 +355,111 @@ async def _handle_start_activity_task( completion.result.completed.result.CopyFrom(payload) except BaseException as err: try: - try: - if isinstance(err, temporalio.activity._CompleteAsyncError): - temporalio.activity.logger.debug("Completing asynchronously") - completion.result.will_complete_async.SetInParent() - elif ( - isinstance( - err, - ( - asyncio.CancelledError, - temporalio.exceptions.CancelledError, - ), - ) - and running_activity.cancelled_due_to_heartbeat_error - ): - err = running_activity.cancelled_due_to_heartbeat_error - temporalio.activity.logger.warning( - f"Completing as failure during heartbeat with error of type {type(err)}: {err}", - ) - await data_converter.encode_failure( - err, completion.result.failed.failure - ) - elif ( - isinstance( - err, - ( - asyncio.CancelledError, - temporalio.exceptions.CancelledError, - ), - ) - and running_activity.cancellation_details.details - and running_activity.cancellation_details.details.paused - ): - temporalio.activity.logger.warning( - "Completing as failure due to unhandled cancel error produced by activity pause", - ) - await data_converter.encode_failure( - temporalio.exceptions.ApplicationError( - type="ActivityPause", - message="Unhandled activity cancel error produced by activity pause", - ), - completion.result.failed.failure, - ) - elif ( - isinstance( - err, - ( - asyncio.CancelledError, - temporalio.exceptions.CancelledError, - ), - ) - and running_activity.cancellation_details.details - and running_activity.cancellation_details.details.reset - ): - temporalio.activity.logger.warning( - "Completing as failure due to unhandled cancel error produced by activity reset", - ) - await data_converter.encode_failure( - temporalio.exceptions.ApplicationError( - type="ActivityReset", - message="Unhandled activity cancel error produced by activity reset", - ), - completion.result.failed.failure, - ) - elif ( + if isinstance(err, temporalio.activity._CompleteAsyncError): + temporalio.activity.logger.debug("Completing asynchronously") + completion.result.will_complete_async.SetInParent() + elif ( + isinstance( + err, + ( + asyncio.CancelledError, + temporalio.exceptions.CancelledError, + ), + ) + and running_activity.cancelled_due_to_heartbeat_error + ): + err = running_activity.cancelled_due_to_heartbeat_error + temporalio.activity.logger.warning( + f"Completing as failure during heartbeat with error of type {type(err)}: {err}", + ) + await data_converter.encode_failure( + err, completion.result.failed.failure + ) + elif ( + isinstance( + err, + ( + asyncio.CancelledError, + temporalio.exceptions.CancelledError, + ), + ) + and running_activity.cancellation_details.details + and running_activity.cancellation_details.details.paused + ): + temporalio.activity.logger.warning( + "Completing as failure due to unhandled cancel error produced by activity pause", + ) + await data_converter.encode_failure( + temporalio.exceptions.ApplicationError( + type="ActivityPause", + message="Unhandled activity cancel error produced by activity pause", + ), + completion.result.failed.failure, + ) + elif ( + isinstance( + err, + ( + asyncio.CancelledError, + temporalio.exceptions.CancelledError, + ), + ) + and running_activity.cancellation_details.details + and running_activity.cancellation_details.details.reset + ): + temporalio.activity.logger.warning( + "Completing as failure due to unhandled cancel error produced by activity reset", + ) + await data_converter.encode_failure( + temporalio.exceptions.ApplicationError( + type="ActivityReset", + message="Unhandled activity cancel error produced by activity reset", + ), + completion.result.failed.failure, + ) + elif ( + isinstance( + err, + ( + asyncio.CancelledError, + temporalio.exceptions.CancelledError, + ), + ) + and running_activity.cancelled_by_request + ): + temporalio.activity.logger.debug("Completing as cancelled") + await data_converter.encode_failure( + # TODO(cretz): Should use some other message? + temporalio.exceptions.CancelledError("Cancelled"), + completion.result.cancelled.failure, + ) + else: + if ( isinstance( err, - ( - asyncio.CancelledError, - temporalio.exceptions.CancelledError, - ), + temporalio.exceptions.ApplicationError, ) - and running_activity.cancelled_by_request + and err.category + == temporalio.exceptions.ApplicationErrorCategory.BENIGN ): - temporalio.activity.logger.debug("Completing as cancelled") - await data_converter.encode_failure( - # TODO(cretz): Should use some other message? - temporalio.exceptions.CancelledError("Cancelled"), - completion.result.cancelled.failure, - ) - elif isinstance( - err, - temporalio.converter._payload_limits._PayloadSizeError, - ): - temporalio.activity.logger.warning( - err.message, - extra={"__temporal_error_identifier": "PayloadSizeError"}, - ) - await data_converter.encode_failure( - err, completion.result.failed.failure + # Downgrade log level to DEBUG for BENIGN application errors. + temporalio.activity.logger.debug( + "Completing activity as failed", + exc_info=True, + extra={"__temporal_error_identifier": "ActivityFailure"}, ) else: - if ( - isinstance( - err, - temporalio.exceptions.ApplicationError, - ) - and err.category - == temporalio.exceptions.ApplicationErrorCategory.BENIGN - ): - # Downgrade log level to DEBUG for BENIGN application errors. - temporalio.activity.logger.debug( - "Completing activity as failed", - exc_info=True, - extra={ - "__temporal_error_identifier": "ActivityFailure" - }, - ) - else: - temporalio.activity.logger.warning( - "Completing activity as failed", - exc_info=True, - extra={ - "__temporal_error_identifier": "ActivityFailure" - }, - ) - await data_converter.encode_failure( - err, completion.result.failed.failure + temporalio.activity.logger.warning( + "Completing activity as failed", + exc_info=True, + extra={"__temporal_error_identifier": "ActivityFailure"}, ) - # For broken executors, we have to fail the entire worker - if isinstance(err, concurrent.futures.BrokenExecutor): - self._fail_worker_exception_queue.put_nowait(err) - # Handle PayloadSizeError from attempting to encode failure information - except ( - temporalio.converter._payload_limits._PayloadSizeError - ) as inner_err: - temporalio.activity.logger.exception(inner_err.message) - completion.result.Clear() await data_converter.encode_failure( - inner_err, completion.result.failed.failure + err, completion.result.failed.failure ) + # For broken executors, we have to fail the entire worker + if isinstance(err, concurrent.futures.BrokenExecutor): + self._fail_worker_exception_queue.put_nowait(err) except Exception as inner_err: temporalio.activity.logger.exception( f"Exception handling failed, original error: {err}" diff --git a/temporalio/worker/_nexus.py b/temporalio/worker/_nexus.py index f35d10fd5..08ecd2f81 100644 --- a/temporalio/worker/_nexus.py +++ b/temporalio/worker/_nexus.py @@ -29,7 +29,6 @@ import temporalio.client import temporalio.common import temporalio.converter -import temporalio.converter._payload_limits import temporalio.nexus from temporalio.bridge.worker import PollShutdownError from temporalio.exceptions import ( @@ -96,15 +95,8 @@ def __init__( self._fail_worker_exception_queue: asyncio.Queue[Exception] = asyncio.Queue() self._worker_shutdown_event: temporalio.common._CompositeEvent | None = None - async def run( - self, - payload_error_limits: temporalio.converter._payload_limits._ServerPayloadErrorLimits - | None, - ) -> None: + async def run(self) -> None: """Continually poll for Nexus tasks and dispatch to handlers.""" - self._data_converter = self._data_converter._with_payload_error_limits( - payload_error_limits - ) async def raise_from_exception_queue() -> NoReturn: raise await self._fail_worker_exception_queue.get() diff --git a/temporalio/worker/_replayer.py b/temporalio/worker/_replayer.py index 508d5f708..dec774ac0 100644 --- a/temporalio/worker/_replayer.py +++ b/temporalio/worker/_replayer.py @@ -307,6 +307,7 @@ def on_eviction_hook( ), nonsticky_to_sticky_poll_ratio=1, no_remote_activities=True, + disable_payload_error_limit=True, task_types=temporalio.bridge.worker.WorkerTaskTypes( enable_workflows=True, enable_local_activities=False, @@ -339,7 +340,7 @@ def on_eviction_hook( bridge_worker_scope = bridge_worker # Start worker - workflow_worker_task = asyncio.create_task(workflow_worker.run(None)) + workflow_worker_task = asyncio.create_task(workflow_worker.run()) # Yield iterator async def replay_iterator() -> AsyncIterator[WorkflowReplayResult]: diff --git a/temporalio/worker/_worker.py b/temporalio/worker/_worker.py index 2ad1d42c6..3a8b216ac 100644 --- a/temporalio/worker/_worker.py +++ b/temporalio/worker/_worker.py @@ -29,7 +29,6 @@ VersioningBehavior, WorkerDeploymentVersion, ) -from temporalio.converter._payload_limits import _ServerPayloadErrorLimits from ._activity import SharedStateManager, _ActivityWorker from ._interceptor import Interceptor @@ -666,6 +665,9 @@ def check_activity(activity: str): ]._to_bridge(), # type: ignore[reportTypedDictNotRequiredAccess,reportOptionalMemberAccess] plugins=deduped_plugin_names, storage_drivers=deduped_storage_driver_types, + disable_payload_error_limit=config.get( + "disable_payload_error_limit", False + ), ), ) @@ -767,17 +769,8 @@ def make_lambda(plugin: Plugin, next: Callable[[Worker], Awaitable[None]]): await next_function(self) async def _run(self): - # Eagerly validate which will do a namespace check in Core - namespace_info = await self._bridge_worker.validate() - payload_error_limits = ( - _ServerPayloadErrorLimits( - 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 - ) + # Eagerly validate which will do a namespace check in Core. + await self._bridge_worker.validate() if self._started: raise RuntimeError("Already started") @@ -797,16 +790,14 @@ async def raise_on_shutdown(): # Create tasks for workers if self._activity_worker: tasks[self._activity_worker] = asyncio.create_task( - self._activity_worker.run(payload_error_limits) + self._activity_worker.run() ) if self._workflow_worker: tasks[self._workflow_worker] = asyncio.create_task( - self._workflow_worker.run(payload_error_limits) + self._workflow_worker.run() ) if self._nexus_worker: - tasks[self._nexus_worker] = asyncio.create_task( - self._nexus_worker.run(payload_error_limits) - ) + tasks[self._nexus_worker] = asyncio.create_task(self._nexus_worker.run()) # Wait for either worker or shutdown requested wait_task = asyncio.wait(tasks.values(), return_when=asyncio.FIRST_EXCEPTION) diff --git a/temporalio/worker/_workflow.py b/temporalio/worker/_workflow.py index 8e6ba2726..f8a135bee 100644 --- a/temporalio/worker/_workflow.py +++ b/temporalio/worker/_workflow.py @@ -23,10 +23,8 @@ import temporalio.common import temporalio.converter import temporalio.converter._extstore -import temporalio.converter._payload_limits import temporalio.exceptions import temporalio.workflow -from temporalio.api.enums.v1 import WorkflowTaskFailedCause from temporalio.bridge.worker import PollShutdownError from temporalio.converter import StorageDriverStoreContext, StorageDriverWorkflowInfo from temporalio.worker.workflow_sandbox._runner import SandboxedWorkflowRunner @@ -206,15 +204,7 @@ def __init__( else: self._dynamic_workflow = defn - async def run( - self, - payload_error_limits: temporalio.converter._payload_limits._ServerPayloadErrorLimits - | None, - ) -> None: - self._data_converter = self._data_converter._with_payload_error_limits( - payload_error_limits - ) - + async def run(self) -> None: # Continually poll for workflow work task_tag = object() try: @@ -486,18 +476,12 @@ async def _handle_activation( upload_metrics = temporalio.converter._extstore.StorageOperationMetrics() try: - try: - upload_metrics = await temporalio.bridge.worker.encode_completion( - completion, - data_converter, - encode_headers=self._encode_headers, - storage_concurrency_limit=self._max_workflow_task_external_storage_concurrency, - ) - except temporalio.converter._payload_limits._PayloadSizeError as err: - logger.warning(err.message) - completion.failed.Clear() - await data_converter.encode_failure(err, completion.failed.failure) - completion.failed.force_cause = WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_PAYLOADS_TOO_LARGE + upload_metrics = await temporalio.bridge.worker.encode_completion( + completion, + data_converter, + encode_headers=self._encode_headers, + storage_concurrency_limit=self._max_workflow_task_external_storage_concurrency, + ) except Exception as err: logger.exception( "Failed encoding completion on workflow with run ID %s", act.run_id @@ -1020,12 +1004,6 @@ async def _decode_payload_sequence( ) -> list[temporalio.api.common.v1.Payload]: return await self._get_current_dc()._decode_payload_sequence(payloads) - def _validate_payload_limits( - self, - payloads: Sequence[temporalio.api.common.v1.Payload], - ) -> None: - self._get_current_dc()._validate_payload_limits(payloads) - class _InterruptDeadlockError(BaseException): pass diff --git a/tests/worker/test_payload_size_limits.py b/tests/worker/test_payload_size_limits.py index 202899c38..510ff12da 100644 --- a/tests/worker/test_payload_size_limits.py +++ b/tests/worker/test_payload_size_limits.py @@ -1,20 +1,14 @@ -import dataclasses import logging import uuid -import warnings from dataclasses import dataclass from datetime import timedelta import pytest -import temporalio -import temporalio.converter from temporalio import activity, workflow from temporalio.client import Client, WorkflowFailureError -from temporalio.converter import PayloadLimitsConfig, PayloadSizeWarning +from temporalio.converter import DataConverter, PayloadLimitsConfig from temporalio.exceptions import ( - ActivityError, - ApplicationError, TerminatedError, TimeoutError, TimeoutType, @@ -27,62 +21,42 @@ TelemetryFilter, ) from temporalio.testing._workflow import WorkflowEnvironment -from temporalio.worker._replayer import Replayer from tests import DEV_SERVER_DOWNLOAD_VERSION from tests.helpers import LogCapturer, new_worker +# Payload/memo size-limit enforcement lives in sdk-rust. These tests only assert that the SDK's +# plumbing reaches core: oversized completions are failed proactively, the worker opt-out lets +# oversized payloads through to the server, and the connection's warn threshold produces a +# forwarded [TMPRL1103] warning. + @dataclass class LargePayloadWorkflowInput: activity_input_data_size: int - activity_output_data_size: int - activity_exception_data_size: int workflow_output_data_size: int - data: str - - -@dataclass -class LargePayloadWorkflowOutput: - data: str @dataclass class LargePayloadActivityInput: - exception_data_size: int - output_data_size: int - data: str - - -@dataclass -class LargePayloadActivityOutput: data: str @activity.defn -async def large_payload_activity( - input: LargePayloadActivityInput, -) -> LargePayloadActivityOutput: - if input.exception_data_size > 0: - raise ApplicationError( - "Intentional activity failure", "e" * input.exception_data_size - ) - return LargePayloadActivityOutput(data="o" * input.output_data_size) +async def large_payload_activity(_input: LargePayloadActivityInput) -> None: + return None @workflow.defn class LargePayloadWorkflow: @workflow.run - async def run(self, input: LargePayloadWorkflowInput) -> LargePayloadWorkflowOutput: - await workflow.execute_activity( - large_payload_activity, - LargePayloadActivityInput( - exception_data_size=input.activity_exception_data_size, - output_data_size=input.activity_output_data_size, - data="i" * input.activity_input_data_size, - ), - schedule_to_close_timeout=timedelta(seconds=5), - ) - return LargePayloadWorkflowOutput(data="o" * input.workflow_output_data_size) + async def run(self, input: LargePayloadWorkflowInput) -> str: + if input.activity_input_data_size > 0: + await workflow.execute_activity( + large_payload_activity, + LargePayloadActivityInput(data="i" * input.activity_input_data_size), + schedule_to_close_timeout=timedelta(seconds=5), + ) + return "o" * input.workflow_output_data_size PAYLOAD_ERROR_LIMIT = 10 * 1024 @@ -95,80 +69,19 @@ async def run(self, input: LargePayloadWorkflowInput) -> LargePayloadWorkflowOut ] -async def test_payload_size_warning_workflow_input(client: Client): - config = client.config() - config["data_converter"] = dataclasses.replace( - temporalio.converter.default(), - payload_limits=PayloadLimitsConfig( - payload_size_warning=100, - ), - ) - client = Client(**config) - - with warnings.catch_warnings(record=True) as w: - async with new_worker( - client, LargePayloadWorkflow, activities=[large_payload_activity] - ) 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=0, - data="i" * 2 * 1024, - ), - id=f"workflow-{uuid.uuid4()}", - task_queue=worker.task_queue, +def _forwarding_runtime(logger: logging.Logger) -> Runtime: + return Runtime( + telemetry=TelemetryConfig( + logging=LoggingConfig( + filter=TelemetryFilter(core_level="WARN", other_level="ERROR"), + forwarding=LogForwardingConfig(logger=logger), ) - - assert len(w) == 1 - assert issubclass(w[-1].category, PayloadSizeWarning) - assert ( - "[TMPRL1103] Attempted to upload payloads with size that exceeded the warning limit." - in str(w[-1].message) ) - - -async def test_payload_size_warning_workflow_memo(client: Client): - config = client.config() - config["data_converter"] = dataclasses.replace( - temporalio.converter.default(), - payload_limits=PayloadLimitsConfig(memo_size_warning=128), ) - client = Client(**config) - - with warnings.catch_warnings(record=True) as w: - async with new_worker( - client, LargePayloadWorkflow, activities=[large_payload_activity] - ) 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=0, - data="", - ), - id=f"workflow-{uuid.uuid4()}", - task_queue=worker.task_queue, - memo={ - "key1": [0] * 64, - "key2": [0] * 64, - "key3": [0] * 64, - }, - ) - - assert len(w) == 1 - 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_payload_size_error_disabled_workflow_payload(env: WorkflowEnvironment): +async def test_oversized_payload_fails_task_with_error_log(env: WorkflowEnvironment): + """An oversized workflow completion is proactively failed by worker, which logs [TMPRL1103].""" if env.supports_time_skipping: pytest.skip("Time-skipping server does not report payload limits.") @@ -176,64 +89,14 @@ async def test_payload_size_error_disabled_workflow_payload(env: WorkflowEnviron dev_server_extra_args=PAYLOAD_LIMITS_EXTRA_ARGS, dev_server_download_version=DEV_SERVER_DOWNLOAD_VERSION, ) as env: - async with new_worker( - env.client, - LargePayloadWorkflow, - activities=[large_payload_activity], - disable_payload_error_limit=True, - ) as worker: - with pytest.raises(WorkflowFailureError) as err: - await env.client.execute_workflow( - LargePayloadWorkflow.run, - LargePayloadWorkflowInput( - activity_input_data_size=PAYLOAD_ERROR_LIMIT + 1024, - activity_output_data_size=0, - activity_exception_data_size=0, - workflow_output_data_size=0, - data="", - ), - id=f"workflow-{uuid.uuid4()}", - task_queue=worker.task_queue, - execution_timeout=timedelta(seconds=3), - ) - - assert isinstance(err.value.cause, TerminatedError) - assert ( - err.value.cause.message - == "BadScheduleActivityAttributes: ScheduleActivityTaskCommandAttributes.Input exceeds size limit." - ) - - -async def test_payload_size_error_workflow_result(env: WorkflowEnvironment): - if env.supports_time_skipping: - pytest.skip("Time-skipping server does not report payload limits.") - - async with await WorkflowEnvironment.start_local( - dev_server_extra_args=PAYLOAD_LIMITS_EXTRA_ARGS, - dev_server_download_version=DEV_SERVER_DOWNLOAD_VERSION, - ) as env: - # Create worker runtime with forwarded logger worker_logger = logging.getLogger(f"log-{uuid.uuid4()}") - worker_runtime = Runtime( - telemetry=TelemetryConfig( - logging=LoggingConfig( - filter=TelemetryFilter(core_level="WARN", other_level="ERROR"), - forwarding=LogForwardingConfig(logger=worker_logger), - ) - ) - ) - - # Create client for worker with custom runtime logging worker_client = await Client.connect( env.client.service_client.config.target_host, namespace=env.client.namespace, - runtime=worker_runtime, + runtime=_forwarding_runtime(worker_logger), ) - with ( - LogCapturer().logs_captured(worker_logger) as worker_logger_capturer, - LogCapturer().logs_captured(logging.getLogger()) as root_logger_capturer, - ): + with LogCapturer().logs_captured(worker_logger) as capturer: async with new_worker( worker_client, LargePayloadWorkflow, activities=[large_payload_activity] ) as worker: @@ -241,10 +104,7 @@ async def test_payload_size_error_workflow_result(env: WorkflowEnvironment): LargePayloadWorkflow.run, LargePayloadWorkflowInput( activity_input_data_size=0, - activity_output_data_size=0, - activity_exception_data_size=0, workflow_output_data_size=PAYLOAD_ERROR_LIMIT + 1024, - data="", ), id=f"workflow-{uuid.uuid4()}", task_queue=worker.task_queue, @@ -253,69 +113,22 @@ async def test_payload_size_error_workflow_result(env: WorkflowEnvironment): with pytest.raises(WorkflowFailureError) as err: await handle.result() - assert isinstance(err.value.cause, TimeoutError) assert err.value.cause.type == TimeoutType.START_TO_CLOSE - replayer = Replayer(workflows=[LargePayloadWorkflow]) - await replayer.replay_workflow(await handle.fetch_history()) - - def worker_logger_predicate(record: logging.LogRecord) -> bool: - return ( - record.levelname == "WARNING" - and "[TMPRL1103] Attempted to upload payloads with size that exceeded the error limit." - in record.msg - ) - - assert worker_logger_capturer.find(worker_logger_predicate) - - def root_logger_predicate(record: logging.LogRecord) -> bool: - return ( - record.levelname == "WARNING" - and "[TMPRL1103] Attempted to upload payloads with size that exceeded the error limit." - in record.msg - ) - - assert root_logger_capturer.find(root_logger_predicate) - - -async def test_payload_size_warning_workflow_result(client: Client): - config = client.config() - config["data_converter"] = dataclasses.replace( - temporalio.converter.default(), - payload_limits=PayloadLimitsConfig( - payload_size_warning=1024, - ), - ) - worker_client = Client(**config) - - with warnings.catch_warnings(record=True) as w: - async with new_worker( - worker_client, LargePayloadWorkflow, activities=[large_payload_activity] - ) 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=2 * 1024, - data="", - ), - id=f"workflow-{uuid.uuid4()}", - task_queue=worker.task_queue, - execution_timeout=timedelta(seconds=3), + def predicate(record: logging.LogRecord) -> bool: + return ( + record.levelname == "ERROR" + and "[TMPRL1103] Attempted to upload payloads with size that exceeded the error limit." + in record.msg ) - assert len(w) == 1 - assert issubclass(w[-1].category, PayloadSizeWarning) - assert ( - "[TMPRL1103] Attempted to upload payloads with size that exceeded the warning limit." - in str(w[-1].message) - ) + assert capturer.find(predicate) -async def test_payload_size_error_activity_input(env: WorkflowEnvironment): +async def test_disable_payload_error_limit_sends_to_server(env: WorkflowEnvironment): + """With the opt-out, worker does not pre-fail; the oversized payload reaches (and is rejected by) + the server.""" if env.supports_time_skipping: pytest.skip("Time-skipping server does not report payload limits.") @@ -323,273 +136,63 @@ async def test_payload_size_error_activity_input(env: WorkflowEnvironment): dev_server_extra_args=PAYLOAD_LIMITS_EXTRA_ARGS, dev_server_download_version=DEV_SERVER_DOWNLOAD_VERSION, ) as env: - # Create worker runtime with forwarded logger - worker_logger = logging.getLogger(f"log-{uuid.uuid4()}") - worker_runtime = Runtime( - telemetry=TelemetryConfig( - logging=LoggingConfig( - filter=TelemetryFilter(core_level="WARN", other_level="ERROR"), - forwarding=LogForwardingConfig(logger=worker_logger), - ) - ) - ) - - # Create client for worker with custom runtime logging - worker_client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, - runtime=worker_runtime, - ) - - with ( - LogCapturer().logs_captured(worker_logger) as worker_logger_capturer, - LogCapturer().logs_captured(logging.getLogger()) as root_logger_capturer, - ): - async with new_worker( - worker_client, LargePayloadWorkflow, activities=[large_payload_activity] - ) as worker: - handle = await env.client.start_workflow( - LargePayloadWorkflow.run, - LargePayloadWorkflowInput( - activity_input_data_size=PAYLOAD_ERROR_LIMIT + 1024, - activity_output_data_size=0, - activity_exception_data_size=0, - workflow_output_data_size=0, - data="", - ), - id=f"workflow-{uuid.uuid4()}", - task_queue=worker.task_queue, - execution_timeout=timedelta(seconds=3), - ) - - with pytest.raises(WorkflowFailureError) as err: - await handle.result() - - assert isinstance(err.value.cause, TimeoutError) - - replayer = Replayer(workflows=[LargePayloadWorkflow]) - await replayer.replay_workflow(await handle.fetch_history()) - - def worker_logger_predicate(record: logging.LogRecord) -> bool: - return ( - record.levelname == "WARNING" - and "[TMPRL1103] Attempted to upload payloads with size that exceeded the error limit." - in record.msg - ) - - assert worker_logger_capturer.find(worker_logger_predicate) - - def root_logger_predicate(record: logging.LogRecord) -> bool: - return ( - record.levelname == "WARNING" - and "[TMPRL1103] Attempted to upload payloads with size that exceeded the error limit." - in record.msg - ) - - assert root_logger_capturer.find(root_logger_predicate) - - -async def test_payload_size_warning_activity_input(client: Client): - config = client.config() - config["data_converter"] = dataclasses.replace( - temporalio.converter.default(), - payload_limits=PayloadLimitsConfig( - payload_size_warning=1024, - ), - ) - worker_client = Client(**config) - - with warnings.catch_warnings(record=True) as w: async with new_worker( - worker_client, LargePayloadWorkflow, activities=[large_payload_activity] + env.client, + LargePayloadWorkflow, + activities=[large_payload_activity], + disable_payload_error_limit=True, ) as worker: - await client.execute_workflow( - LargePayloadWorkflow.run, - LargePayloadWorkflowInput( - activity_input_data_size=2 * 1024, - activity_output_data_size=0, - activity_exception_data_size=0, - workflow_output_data_size=0, - data="", - ), - id=f"workflow-{uuid.uuid4()}", - task_queue=worker.task_queue, - ) - - assert len(w) == 1 - assert issubclass(w[-1].category, PayloadSizeWarning) - assert ( - "[TMPRL1103] Attempted to upload payloads with size that exceeded the warning limit." - in str(w[-1].message) - ) - - -async def test_payload_size_error_activity_exception(env: WorkflowEnvironment): - if env.supports_time_skipping: - pytest.skip("Time-skipping server does not report payload limits.") - - async with await WorkflowEnvironment.start_local( - dev_server_extra_args=PAYLOAD_LIMITS_EXTRA_ARGS, - dev_server_download_version=DEV_SERVER_DOWNLOAD_VERSION, - ) as env: - # Create worker runtime with forwarded logger - worker_logger = logging.getLogger(f"log-{uuid.uuid4()}") - worker_runtime = Runtime( - telemetry=TelemetryConfig( - logging=LoggingConfig( - filter=TelemetryFilter(core_level="WARN", other_level="ERROR"), - forwarding=LogForwardingConfig(logger=worker_logger), - ) - ) - ) - - # Create client for worker with custom runtime logging - worker_client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, - runtime=worker_runtime, - ) - - with ( - LogCapturer().logs_captured( - activity.logger.base_logger - ) as activity_logger_capturer, - ): - async with new_worker( - worker_client, LargePayloadWorkflow, activities=[large_payload_activity] - ) as worker: - handle = await env.client.start_workflow( + with pytest.raises(WorkflowFailureError) as err: + await env.client.execute_workflow( LargePayloadWorkflow.run, LargePayloadWorkflowInput( - activity_input_data_size=0, - activity_output_data_size=0, - activity_exception_data_size=PAYLOAD_ERROR_LIMIT + 1024, + activity_input_data_size=PAYLOAD_ERROR_LIMIT + 1024, workflow_output_data_size=0, - data="", ), id=f"workflow-{uuid.uuid4()}", task_queue=worker.task_queue, + execution_timeout=timedelta(seconds=3), ) - with pytest.raises(WorkflowFailureError) as err: - await handle.result() - - assert isinstance(err.value.cause, ActivityError) - assert isinstance(err.value.cause.cause, ApplicationError) - - replayer = Replayer(workflows=[LargePayloadWorkflow]) - await replayer.replay_workflow(await handle.fetch_history()) - - def activity_logger_predicate(record: logging.LogRecord) -> bool: - return ( - record.levelname == "ERROR" - and "[TMPRL1103] Attempted to upload payloads with size that exceeded the error limit." - in record.msg - ) - - assert activity_logger_capturer.find(activity_logger_predicate) - - -async def test_payload_size_error_activity_result(env: WorkflowEnvironment): - if env.supports_time_skipping: - pytest.skip("Time-skipping server does not report payload limits.") - - async with await WorkflowEnvironment.start_local( - dev_server_extra_args=PAYLOAD_LIMITS_EXTRA_ARGS, - dev_server_download_version=DEV_SERVER_DOWNLOAD_VERSION, - ) as env: - # Create worker runtime with forwarded logger - worker_logger = logging.getLogger(f"log-{uuid.uuid4()}") - worker_runtime = Runtime( - telemetry=TelemetryConfig( - logging=LoggingConfig( - filter=TelemetryFilter(core_level="WARN", other_level="ERROR"), - forwarding=LogForwardingConfig(logger=worker_logger), - ) + assert isinstance(err.value.cause, TerminatedError) + assert ( + err.value.cause.message + == "BadScheduleActivityAttributes: ScheduleActivityTaskCommandAttributes.Input exceeds size limit." ) - ) - # Create client for worker with custom runtime logging - worker_client = await Client.connect( - env.client.service_client.config.target_host, - namespace=env.client.namespace, - runtime=worker_runtime, - ) - - with ( - LogCapturer().logs_captured( - activity.logger.base_logger - ) as activity_logger_capturer, - ): - async with new_worker( - worker_client, LargePayloadWorkflow, activities=[large_payload_activity] - ) as worker: - handle = await env.client.start_workflow( - LargePayloadWorkflow.run, - LargePayloadWorkflowInput( - activity_input_data_size=0, - activity_output_data_size=PAYLOAD_ERROR_LIMIT + 1024, - activity_exception_data_size=0, - workflow_output_data_size=0, - data="", - ), - id=f"workflow-{uuid.uuid4()}", - task_queue=worker.task_queue, - ) - - with pytest.raises(WorkflowFailureError) as err: - await handle.result() - - assert isinstance(err.value.cause, ActivityError) - assert isinstance(err.value.cause.cause, ApplicationError) - assert handle is not None - replayer = Replayer(workflows=[LargePayloadWorkflow]) - await replayer.replay_workflow(await handle.fetch_history()) - - def activity_logger_predicate(record: logging.LogRecord) -> bool: - return ( - hasattr(record, "__temporal_error_identifier") - and getattr(record, "__temporal_error_identifier") - == "PayloadSizeError" - and record.levelname == "WARNING" - and "[TMPRL1103] Attempted to upload payloads with size that exceeded the error limit." - in record.msg - ) - - assert activity_logger_capturer.find(activity_logger_predicate) - - -async def test_payload_size_warning_activity_result(client: Client): - config = client.config() - config["data_converter"] = dataclasses.replace( - temporalio.converter.default(), - payload_limits=PayloadLimitsConfig( - payload_size_warning=1024, +async def test_payload_size_warning_forwarded(env: WorkflowEnvironment): + """The connection's warn threshold produces a forwarded [TMPRL1103] warning for over-threshold payloads.""" + worker_logger = logging.getLogger(f"log-{uuid.uuid4()}") + worker_client = await Client.connect( + env.client.service_client.config.target_host, + namespace=env.client.namespace, + runtime=_forwarding_runtime(worker_logger), + data_converter=DataConverter( + payload_limits=PayloadLimitsConfig(payload_size_warning=1024), ), ) - worker_client = Client(**config) - with warnings.catch_warnings(record=True) as w: + with LogCapturer().logs_captured(worker_logger) as capturer: async with new_worker( worker_client, LargePayloadWorkflow, activities=[large_payload_activity] ) as worker: - await client.execute_workflow( + await worker_client.execute_workflow( LargePayloadWorkflow.run, LargePayloadWorkflowInput( activity_input_data_size=0, - activity_output_data_size=2 * 1024, - activity_exception_data_size=0, - workflow_output_data_size=0, - data="", + workflow_output_data_size=2 * 1024, ), id=f"workflow-{uuid.uuid4()}", task_queue=worker.task_queue, + execution_timeout=timedelta(seconds=5), ) - assert len(w) == 1 - assert issubclass(w[-1].category, PayloadSizeWarning) - assert ( - "[TMPRL1103] Attempted to upload payloads with size that exceeded the warning limit." - in str(w[-1].message) + def predicate(record: logging.LogRecord) -> bool: + return ( + record.levelname == "WARNING" + and "[TMPRL1103] Attempted to upload payloads with size that exceeded the warning limit." + in record.msg ) + + assert capturer.find(predicate)