💥 feature: payloads/memo size enforcement

CHANGELOG.md

@@ -19,6 +19,19 @@ to docs, or any other relevant information.
 
 ## [Unreleased]
 
+### Breaking Changes
+
+- Payload/memo size-limit enforcement (experimental) is now on by default. Workers proactively
+  validate outbound payload/memo sizes before sending: a field over the warn threshold is logged
+  (`[TMPRL1103]` at `WARN`) but still sent, while a task completion over the error limit is failed
+  retryably (`[TMPRL1103]` at `ERROR`) instead of sent. Previously these reached the server, which
+  terminated the workflow or failed the activity non-retryably; failing retryably instead lets a
+  corrected workflow or activity be redeployed and recover. Tune warn thresholds via
+  `Temporalio::Client::Connection::PayloadLimitsOptions` (passed as the connection's
+  `payload_limits:`). If you use a proxy between the worker and server that alters the size of
+  payloads (e.g. compression, encryption, external storage), it is advised that you disable size
+  enforcement by setting `disable_payload_error_limit: true` on the worker.
+
 ## [v1.6.0] - 2026-07-16
 
 ### Added

temporalio/ext/src/client.rs

@@ -3,7 +3,7 @@ use std::{collections::HashMap, future::Future, marker::PhantomData, time::Durat
 use temporalio_client::{
     ClientKeepAliveOptions, ClientTlsOptions, Connection, ConnectionOptions,
     DnsLoadBalancingOptions, GrpcCompression as CoreGrpcCompression, HttpConnectProxyOptions,
-    RetryOptions, TlsOptions, errors::ClientConnectError,
+    PayloadLimitsOptions, RetryOptions, TlsOptions, errors::ClientConnectError,
 };
 
 use magnus::{
@@ -206,6 +206,12 @@ impl Client {
         } else {
             None
         })
+        // The connection layer always supplies concrete thresholds (512 KiB / 2 KiB by default);
+        // `0` disables that warning.
+        .payload_limits(PayloadLimitsOptions {
+            payloads_warn_size: options.member::<u64>(id!("payloads_warn_size"))?,
+            memo_warn_size: options.member::<u64>(id!("memo_warn_size"))?,
+        })
         .maybe_metrics_meter(metrics_meter)
         .build();
 

temporalio/ext/src/worker.rs

@@ -39,10 +39,11 @@ use temporalio_common::{
     worker::{WorkerDeploymentOptions, WorkerDeploymentVersion, WorkerTaskTypes},
 };
 use temporalio_sdk_core::{
-    PollError, PollerBehavior, ResourceBasedSlotsOptions, ResourceSlotOptions, SlotInfo,
-    SlotInfoTrait, SlotKind, SlotKindType, SlotMarkUsedContext, SlotReleaseContext,
-    SlotReservationContext, SlotSupplier, SlotSupplierOptions, SlotSupplierPermit, TunerHolder,
-    TunerHolderOptions, WorkerConfig, WorkerVersioningStrategy, WorkflowErrorType,
+    PollError, PollerBehavior, ResourceBasedSlotsOptions, ResourceBasedTunerConfig,
+    ResourceSlotOptions, SlotInfo, SlotInfoTrait, SlotKind, SlotKindType, SlotMarkUsedContext,
+    SlotReleaseContext, SlotReservationContext, SlotSupplier, SlotSupplierOptions,
+    SlotSupplierPermit, TunerHolder, TunerHolderOptions, WorkerConfig, WorkerVersioningStrategy,
+    WorkflowErrorType,
     replay::{HistoryForReplay, ReplayWorkerInput},
 };
 use tokio::sync::mpsc::{Sender, UnboundedSender, channel, unbounded_channel};
@@ -576,6 +577,7 @@ fn build_config(options: Struct, runtime_handle: &RuntimeHandle) -> Result<Worke
                 })
                 .collect::<HashSet<_>>(),
         )
+        .disable_payload_error_limit(options.member::<bool>(id!("disable_payload_error_limit"))?)
         .build()
         .map_err(|err| error!("Invalid worker options: {}", err))
 }
@@ -616,7 +618,7 @@ fn build_tuner(options: Struct, runtime_handle: &RuntimeHandle) -> Result<TunerH
     )?;
 
     TunerHolderOptions::builder()
-        .maybe_resource_based_options(resource_slot_options)
+        .maybe_resource_based_config(resource_slot_options.map(ResourceBasedTunerConfig::Options))
         .workflow_slot_options(workflow_slot_options)
         .activity_slot_options(activity_slot_options)
         .local_activity_slot_options(local_activity_slot_options)

temporalio/lib/temporalio/client/connection.rb

@@ -26,7 +26,8 @@ class Connection
         :runtime,
         :lazy_connect,
         :dns_load_balancing,
-        :grpc_compression
+        :grpc_compression,
+        :payload_limits
       )
 
       # Options as returned from {options} for +**to_h+ splat use in {initialize}. See {initialize} for details.
@@ -169,6 +170,28 @@ def codec
         end
       end
 
+      PayloadLimitsOptions = Data.define(
+        :payloads_warn_size,
+        :memo_warn_size
+      )
+
+      # Payload size-limit options for client connections.
+      #
+      # @note WARNING: Payload size-limit enforcement is experimental and the API may change in the future.
+      #
+      # @!attribute payloads_warn_size
+      #   @return [Integer] Warning threshold, in bytes, for the size of an outbound payload-bearing field.
+      #     Over-threshold fields are logged but still sent to the server. Defaults to 512 KiB; +0+ disables the
+      #     warning.
+      # @!attribute memo_warn_size
+      #   @return [Integer] Warning threshold, in bytes, for outbound memo size. Over-threshold memos are logged
+      #     but still sent to the server. Defaults to 2 KiB; +0+ disables the warning.
+      class PayloadLimitsOptions
+        def initialize(payloads_warn_size: 512 * 1024, memo_warn_size: 2 * 1024)
+          super
+        end
+      end
+
       # @return [Options] Frozen options for this client which has the same attributes as {initialize}. Note that if
       #   {api_key=} or {rpc_metadata=} are updated, the options object is replaced with those changes (it is not
       #   mutated in place).
@@ -210,6 +233,9 @@ def codec
       #   is +nil+ (disabled). Silently disabled when +http_connect_proxy+ is set, since the two are mutually exclusive.
       # @param grpc_compression [GrpcCompressionOptions::Gzip, GrpcCompressionOptions::None] Transport-level gRPC
       #   compression. Defaults to gzip. Set to {GrpcCompressionOptions::None} to opt out.
+      # @param payload_limits [PayloadLimitsOptions] Payload size-limit options for this connection. Defaults to a
+      #   {PayloadLimitsOptions} with the standard warning thresholds; see that type for the defaults and how to disable
+      #   a warning. WARNING: This is experimental and may change in the future.
       # @param around_connect [Proc, nil] If present, this proc accepts two values: options and a block. The block must
       #   be yielded to only once with the options. The block does not return a meaningful value, nor should
       #   around_connect.
@@ -228,6 +254,7 @@ def initialize(
         lazy_connect: false,
         dns_load_balancing: nil,
         grpc_compression: GrpcCompressionOptions::Gzip.new,
+        payload_limits: PayloadLimitsOptions.new,
         around_connect: nil
       )
         @options = Options.new(
@@ -242,7 +269,8 @@ def initialize(
           runtime:,
           lazy_connect:,
           dns_load_balancing:,
-          grpc_compression:
+          grpc_compression:,
+          payload_limits:
         ).freeze
         @core_client_mutex = Mutex.new
         # Create core client now if not lazy, applying around_connect if present
@@ -344,7 +372,9 @@ def new_core_client
             max_elapsed_time: @options.rpc_retry.max_elapsed_time,
             max_retries: @options.rpc_retry.max_retries
           ),
-          identity: @options.identity || "#{Process.pid}@#{Socket.gethostname}"
+          identity: @options.identity || "#{Process.pid}@#{Socket.gethostname}",
+          payloads_warn_size: @options.payload_limits.payloads_warn_size,
+          memo_warn_size: @options.payload_limits.memo_warn_size
         )
         grpc_compression_codec = case @options.grpc_compression
                                  when GrpcCompressionOptions::Gzip

temporalio/lib/temporalio/internal/bridge/client.rb

@@ -18,7 +18,9 @@ class Client
           :keep_alive, # Optional
           :http_connect_proxy, # Optional
           :dns_load_balancing, # Optional
-          :grpc_compression
+          :grpc_compression,
+          :payloads_warn_size,
+          :memo_warn_size
         )
 
         TLSOptions = Struct.new(

temporalio/lib/temporalio/internal/bridge/worker.rb

@@ -28,7 +28,8 @@ class Worker
           :nondeterminism_as_workflow_fail,
           :nondeterminism_as_workflow_fail_for_types,
           :deployment_options,
-          :plugins
+          :plugins,
+          :disable_payload_error_limit
         )
 
         TunerOptions = Struct.new(

temporalio/lib/temporalio/runtime.rb

@@ -101,8 +101,8 @@ def initialize(core_level: 'WARN', other_level: 'ERROR')
       # @!visibility private
       def _to_bridge
         # @type self: LoggingFilterOptions
-        "#{other_level},temporalio_sdk_core=#{core_level},temporalio_client=#{core_level}," \
-          "temporalio_sdk=#{core_level},temporalio_bridge=#{core_level}"
+        "#{other_level},temporalio_common=#{core_level},temporalio_sdk_core=#{core_level}," \
+          "temporalio_client=#{core_level},temporalio_sdk=#{core_level},temporalio_bridge=#{core_level}"
       end
     end
 

temporalio/lib/temporalio/worker.rb

@@ -71,7 +71,8 @@ class Worker
       :patch_activation_callback,
       :workflow_task_poller_behavior,
       :activity_task_poller_behavior,
-      :debug_mode
+      :debug_mode,
+      :disable_payload_error_limit
     )
 
     # Options as returned from {options} for `**to_h` splat use in {initialize}. See {initialize} for details.
@@ -452,6 +453,10 @@ def self._validate_plugins!(plugins)
     # @param debug_mode [Boolean] If true, deadlock detection is disabled. Deadlock detection will fail workflow tasks
     #   if they block the thread for too long. This defaults to true if the `TEMPORAL_DEBUG` environment variable is
     #   `true` or `1`.
+    # @param disable_payload_error_limit [Boolean] If true, the worker will not proactively fail workflow/activity
+    #   tasks whose payloads exceed the namespace error limits; oversized payloads are sent to the server, which
+    #   enforces the limit. Defaults to false (the worker fails such tasks before sending). WARNING: Payload
+    #   size-limit enforcement is experimental and may change in the future.
     def initialize(
       client:,
       task_queue:,
@@ -484,7 +489,8 @@ def initialize(
       patch_activation_callback: nil,
       workflow_task_poller_behavior: PollerBehavior::SimpleMaximum.new(max_concurrent_workflow_task_polls),
       activity_task_poller_behavior: PollerBehavior::SimpleMaximum.new(max_concurrent_activity_task_polls),
-      debug_mode: %w[true 1].include?(ENV['TEMPORAL_DEBUG'].to_s.downcase)
+      debug_mode: %w[true 1].include?(ENV['TEMPORAL_DEBUG'].to_s.downcase),
+      disable_payload_error_limit: false
     )
       Internal::ProtoUtils.assert_non_reserved_name(task_queue)
 
@@ -520,7 +526,8 @@ def initialize(
         patch_activation_callback:,
         workflow_task_poller_behavior:,
         activity_task_poller_behavior:,
-        debug_mode:
+        debug_mode:,
+        disable_payload_error_limit:
       ).freeze
       # Collect applicable client plugins and worker plugins, then validate and apply to options
       @plugins = client.options.plugins.grep(Plugin) + plugins
@@ -581,7 +588,8 @@ def _initialize_from_options
           nondeterminism_as_workflow_fail:,
           nondeterminism_as_workflow_fail_for_types:,
           deployment_options: @options.deployment_options._to_bridge_options,
-          plugins: (@options.client.options.plugins + @options.plugins).map(&:name).uniq.sort
+          plugins: (@options.client.options.plugins + @options.plugins).map(&:name).uniq.sort,
+          disable_payload_error_limit: @options.disable_payload_error_limit
         )
       )
 

temporalio/lib/temporalio/worker/workflow_replayer.rb

@@ -244,6 +244,7 @@ def initialize(
               nondeterminism_as_workflow_fail:,
               nondeterminism_as_workflow_fail_for_types:,
               deployment_options: Worker.default_deployment_options._to_bridge_options,
+              disable_payload_error_limit: true,
               plugins: options.plugins.map(&:name).uniq.sort
             )
           )

temporalio/rbi/temporalio/client/connection.rbi

@@ -15,6 +15,7 @@ class Temporalio::Client::Connection
       lazy_connect: T::Boolean,
       dns_load_balancing: T.nilable(Temporalio::Client::Connection::DnsLoadBalancingOptions),
       grpc_compression: T.any(Temporalio::Client::Connection::GrpcCompressionOptions::Gzip, Temporalio::Client::Connection::GrpcCompressionOptions::None),
+      payload_limits: Temporalio::Client::Connection::PayloadLimitsOptions,
       around_connect: T.nilable(T.proc.params(arg0: Temporalio::Client::Connection::Options, arg1: T.proc.params(arg0: Temporalio::Client::Connection::Options).void).void)
     ).void
   end
@@ -31,6 +32,7 @@ class Temporalio::Client::Connection
     lazy_connect: T.unsafe(nil),
     dns_load_balancing: T.unsafe(nil),
     grpc_compression: T.unsafe(nil),
+    payload_limits: T.unsafe(nil),
     around_connect: T.unsafe(nil)
   ); end
 

temporalio/rbi/temporalio/worker.rbi

@@ -39,7 +39,8 @@ class Temporalio::Worker
       patch_activation_callback: T.nilable(T.proc.params(input: Temporalio::Worker::PatchActivationInput).returns(T::Boolean)),
       workflow_task_poller_behavior: Temporalio::Worker::PollerBehavior,
       activity_task_poller_behavior: Temporalio::Worker::PollerBehavior,
-      debug_mode: T::Boolean
+      debug_mode: T::Boolean,
+      disable_payload_error_limit: T::Boolean
     ).void
   end
   def initialize(
@@ -74,7 +75,8 @@ class Temporalio::Worker
     patch_activation_callback: T.unsafe(nil),
     workflow_task_poller_behavior: T.unsafe(nil),
     activity_task_poller_behavior: T.unsafe(nil),
-    debug_mode: T.unsafe(nil)
+    debug_mode: T.unsafe(nil),
+    disable_payload_error_limit: T.unsafe(nil)
   ); end
 
   sig { returns(String) }

temporalio/sig/temporalio/client/connection.rbs

@@ -14,6 +14,7 @@ module Temporalio
         attr_reader lazy_connect: bool
         attr_reader dns_load_balancing: DnsLoadBalancingOptions?
         attr_reader grpc_compression: GrpcCompressionOptions::Gzip | GrpcCompressionOptions::None
+        attr_reader payload_limits: PayloadLimitsOptions
 
         def initialize: (
           target_host: String,
@@ -27,7 +28,8 @@ module Temporalio
           runtime: Runtime,
           lazy_connect: bool,
           dns_load_balancing: DnsLoadBalancingOptions?,
-          grpc_compression: GrpcCompressionOptions::Gzip | GrpcCompressionOptions::None
+          grpc_compression: GrpcCompressionOptions::Gzip | GrpcCompressionOptions::None,
+          payload_limits: PayloadLimitsOptions
         ) -> void
 
         def to_h: -> Hash[Symbol, untyped]
@@ -107,6 +109,16 @@ module Temporalio
         end
       end
 
+      class PayloadLimitsOptions
+        attr_reader payloads_warn_size: Integer?
+        attr_reader memo_warn_size: Integer?
+
+        def initialize: (
+          ?payloads_warn_size: Integer?,
+          ?memo_warn_size: Integer?
+        ) -> void
+      end
+
       attr_reader options: Options
 
       attr_reader workflow_service: WorkflowService
@@ -126,6 +138,7 @@ module Temporalio
         ?lazy_connect: bool,
         ?dns_load_balancing: DnsLoadBalancingOptions?,
         ?grpc_compression: GrpcCompressionOptions::Gzip | GrpcCompressionOptions::None,
+        ?payload_limits: PayloadLimitsOptions,
         ?around_connect: nil | ^(Options) { (Options) -> void } -> void
       ) -> void
 

temporalio/sig/temporalio/internal/bridge/client.rbs

@@ -15,6 +15,8 @@ module Temporalio
           attr_accessor http_connect_proxy: HTTPConnectProxyOptions?
           attr_accessor dns_load_balancing: DnsLoadBalancingOptions?
           attr_accessor grpc_compression: GrpcCompressionOptions?
+          attr_accessor payloads_warn_size: Integer?
+          attr_accessor memo_warn_size: Integer?
 
           def initialize: (
             target_host: String,
@@ -28,7 +30,9 @@ module Temporalio
             ?keep_alive: KeepAliveOptions?,
             ?http_connect_proxy: HTTPConnectProxyOptions?,
             ?dns_load_balancing: DnsLoadBalancingOptions?,
-            ?grpc_compression: GrpcCompressionOptions?
+            ?grpc_compression: GrpcCompressionOptions?,
+            ?payloads_warn_size: Integer?,
+            ?memo_warn_size: Integer?
           ) -> void
         end
 

temporalio/sig/temporalio/internal/bridge/worker.rbs

@@ -25,6 +25,7 @@ module Temporalio
           attr_accessor nondeterminism_as_workflow_fail_for_types: Array[String]
           attr_accessor deployment_options: DeploymentOptions?
           attr_accessor plugins: Array[String]
+          attr_accessor disable_payload_error_limit: bool
 
           def initialize: (
             namespace: String,
@@ -48,7 +49,8 @@ module Temporalio
             nondeterminism_as_workflow_fail: bool,
             nondeterminism_as_workflow_fail_for_types: Array[String],
             deployment_options: DeploymentOptions?,
-            plugins: Array[String]
+            plugins: Array[String],
+            disable_payload_error_limit: bool
           ) -> void
         end
 

temporalio/sig/temporalio/worker.rbs

@@ -41,6 +41,7 @@ module Temporalio
       attr_reader deployment_options: Worker::DeploymentOptions
       attr_reader patch_activation_callback: Proc?
       attr_reader debug_mode: bool
+      attr_reader disable_payload_error_limit: bool
 
       def initialize: (
         client: Client,
@@ -74,7 +75,8 @@ module Temporalio
         activity_task_poller_behavior: PollerBehavior,
         deployment_options: Worker::DeploymentOptions,
         patch_activation_callback: Proc?,
-        debug_mode: bool
+        debug_mode: bool,
+        disable_payload_error_limit: bool
       ) -> void
 
       def with: (**Object kwargs) -> Options
@@ -138,7 +140,8 @@ module Temporalio
       ?activity_task_poller_behavior: PollerBehavior,
       ?deployment_options: Worker::DeploymentOptions,
       ?patch_activation_callback: Proc?,
-      ?debug_mode: bool
+      ?debug_mode: bool,
+      ?disable_payload_error_limit: bool
     ) -> void
 
     def _initialize_from_options: -> void