diff --git a/CHANGELOG.md b/CHANGELOG.md index 580155bd..363d684a 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/temporalio/ext/sdk-core b/temporalio/ext/sdk-core index 16452094..3dac9013 160000 --- a/temporalio/ext/sdk-core +++ b/temporalio/ext/sdk-core @@ -1 +1 @@ -Subproject commit 16452094827ab38bfb84f5ac67a627d66350d07f +Subproject commit 3dac9013b9031e5ffd51d7335838585b2db42efb diff --git a/temporalio/ext/src/client.rs b/temporalio/ext/src/client.rs index cecb2855..a5420d73 100644 --- a/temporalio/ext/src/client.rs +++ b/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::(id!("payloads_warn_size"))?, + memo_warn_size: options.member::(id!("memo_warn_size"))?, + }) .maybe_metrics_meter(metrics_meter) .build(); diff --git a/temporalio/ext/src/worker.rs b/temporalio/ext/src/worker.rs index 709141cb..7e5749e0 100644 --- a/temporalio/ext/src/worker.rs +++ b/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>(), ) + .disable_payload_error_limit(options.member::(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 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 diff --git a/temporalio/sig/temporalio/internal/bridge/client.rbs b/temporalio/sig/temporalio/internal/bridge/client.rbs index eb9a6415..d9a3e89d 100644 --- a/temporalio/sig/temporalio/internal/bridge/client.rbs +++ b/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 diff --git a/temporalio/sig/temporalio/internal/bridge/worker.rbs b/temporalio/sig/temporalio/internal/bridge/worker.rbs index 51272e33..62139c3e 100644 --- a/temporalio/sig/temporalio/internal/bridge/worker.rbs +++ b/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 diff --git a/temporalio/sig/temporalio/worker.rbs b/temporalio/sig/temporalio/worker.rbs index d8bf4d1d..8560a32c 100644 --- a/temporalio/sig/temporalio/worker.rbs +++ b/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 diff --git a/temporalio/test/sig/worker_payload_size_limits_test.rbs b/temporalio/test/sig/worker_payload_size_limits_test.rbs new file mode 100644 index 00000000..d8ebd33e --- /dev/null +++ b/temporalio/test/sig/worker_payload_size_limits_test.rbs @@ -0,0 +1,13 @@ +class WorkerPayloadSizeLimitsTest < Test + PAYLOAD_ERROR_LIMIT: Integer + + class LargePayloadActivity < Temporalio::Activity::Definition + def execute: (untyped data) -> untyped + end + + class LargePayloadWorkflow < Temporalio::Workflow::Definition + def execute: (untyped activity_input_data_size, untyped workflow_output_data_size) -> untyped + end + + def with_payload_limited_server: [T] { (Temporalio::Testing::WorkflowEnvironment server) -> T } -> T +end diff --git a/temporalio/test/worker_payload_size_limits_test.rb b/temporalio/test/worker_payload_size_limits_test.rb new file mode 100644 index 00000000..19f75e2f --- /dev/null +++ b/temporalio/test/worker_payload_size_limits_test.rb @@ -0,0 +1,114 @@ +# frozen_string_literal: true + +require 'securerandom' +require 'temporalio/client' +require 'temporalio/error' +require 'temporalio/runtime' +require 'temporalio/testing' +require 'temporalio/worker' +require 'temporalio/workflow' +require 'test' + +# Payload/memo size-limit enforcement lives in sdk-core. These tests only assert that the Ruby +# plumbing reaches core: an oversized worker completion is failed proactively, and the +# disable_payload_error_limit opt-out lets the oversized payload reach (and be rejected by) the +# server. The warning channel is asserted via the default log filter. +class WorkerPayloadSizeLimitsTest < Test + PAYLOAD_ERROR_LIMIT = 10 * 1024 + + class LargePayloadActivity < Temporalio::Activity::Definition + def execute(_data) + nil + end + end + + class LargePayloadWorkflow < Temporalio::Workflow::Definition + def execute(activity_input_data_size, workflow_output_data_size) + if activity_input_data_size.positive? + Temporalio::Workflow.execute_activity( + LargePayloadActivity, + 'i' * activity_input_data_size, + schedule_to_close_timeout: 5 + ) + end + 'o' * workflow_output_data_size + end + end + + # Regression guard for the log-filter fix: the default filter must admit temporalio_common (where + # validate_payload_limits logs the [TMPRL1103] warning) at the core level, otherwise payload-limit + # warnings would be silently dropped. + def test_default_log_filter_admits_temporalio_common + filter = Temporalio::Runtime::LoggingFilterOptions.new(core_level: 'WARN', other_level: 'ERROR') + assert_includes filter._to_bridge, 'temporalio_common=WARN' + end + + def with_payload_limited_server + server = Temporalio::Testing::WorkflowEnvironment.start_local( + dev_server_extra_args: [ + '--dynamic-config-value', "limit.blobSize.error=#{PAYLOAD_ERROR_LIMIT}", + # The server only enforces the error limit for payloads that also exceed the warn limit, so + # the warn limit must be below the error limit. + '--dynamic-config-value', 'limit.blobSize.warn=2048' + ] + ) + begin + yield server + ensure + server.shutdown + end + end + + def test_oversized_payload_fails_task + with_payload_limited_server do |server| + worker = Temporalio::Worker.new( + client: server.client, + task_queue: "tq-#{SecureRandom.uuid}", + activities: [LargePayloadActivity], + workflows: [LargePayloadWorkflow] + ) + # Rather than send the oversized workflow result, core proactively fails the workflow task with a + # PAYLOADS_TOO_LARGE cause. The task keeps failing, so the workflow ultimately fails by execution timeout. + worker.run do + handle = server.client.start_workflow( + LargePayloadWorkflow, 0, PAYLOAD_ERROR_LIMIT + 1024, + id: "wf-#{SecureRandom.uuid}", + task_queue: worker.task_queue, + execution_timeout: 3 + ) + assert_raises(Temporalio::Error::WorkflowFailedError) { handle.result } + + # Confirm the driving cause: the worker reported a workflow task failure with PAYLOADS_TOO_LARGE. + # @type var events: Array[untyped] + events = handle.fetch_history_events.to_a + assert(events.any? do |event| + event.event_type == :EVENT_TYPE_WORKFLOW_TASK_FAILED && + event.workflow_task_failed_event_attributes.cause == :WORKFLOW_TASK_FAILED_CAUSE_PAYLOADS_TOO_LARGE + end) + end + end + end + + def test_disable_payload_error_limit_sends_to_server + with_payload_limited_server do |server| + worker = Temporalio::Worker.new( + client: server.client, + task_queue: "tq-#{SecureRandom.uuid}", + activities: [LargePayloadActivity], + workflows: [LargePayloadWorkflow], + disable_payload_error_limit: true + ) + # With the opt-out, core does not pre-fail; the oversized activity input reaches the server, + # which rejects the ScheduleActivityTask command and fails the workflow. + worker.run do + handle = server.client.start_workflow( + LargePayloadWorkflow, PAYLOAD_ERROR_LIMIT + 1024, 0, + id: "wf-#{SecureRandom.uuid}", + task_queue: worker.task_queue, + execution_timeout: 3 + ) + assert_raises(Temporalio::Error::WorkflowFailedError) { handle.result } + end + end + end +end