diff --git a/CHANGELOG.md b/CHANGELOG.md index ec6c1be8a..b1f00f560 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,18 @@ 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 + `NativeConnectionOptions.payloadLimits`. 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 `disablePayloadErrorLimit: true` on the worker. + ## [1.20.3] - 2026-07-13 ### Fixed diff --git a/packages/core-bridge/sdk-core b/packages/core-bridge/sdk-core index afb5251f9..56d137b32 160000 --- a/packages/core-bridge/sdk-core +++ b/packages/core-bridge/sdk-core @@ -1 +1 @@ -Subproject commit afb5251f97b021a43ca4089906f401413956fcb5 +Subproject commit 56d137b321156179bccf179ef57e4e33eb987b18 diff --git a/packages/core-bridge/src/client.rs b/packages/core-bridge/src/client.rs index 71f0e4f2e..08271d416 100644 --- a/packages/core-bridge/src/client.rs +++ b/packages/core-bridge/src/client.rs @@ -622,7 +622,7 @@ mod config { use temporalio_client::{ ClientTlsOptions as CoreClientTlsOptions, ConnectionOptions, DnsLoadBalancingOptions, - GrpcCompression as CoreGrpcCompression, HttpConnectProxyOptions, + GrpcCompression as CoreGrpcCompression, HttpConnectProxyOptions, PayloadLimitsOptions, TlsOptions as CoreTlsOptions, }; use temporalio_common::telemetry::metrics::TemporalMeter; @@ -648,6 +648,8 @@ mod config { headers: Option>, api_key: Option, disable_error_code_metric_tags: bool, + payloads_warn_size: u64, + memo_warn_size: u64, } #[derive(Debug, Clone, TryFromJs)] @@ -721,6 +723,10 @@ mod config { .maybe_api_key(self.api_key) .maybe_metrics_meter(metrics_meter) .disable_error_code_metric_tags(self.disable_error_code_metric_tags) + .payload_limits(PayloadLimitsOptions { + payloads_warn_size: self.payloads_warn_size, + memo_warn_size: self.memo_warn_size, + }) // identity -- skipped: will be set on worker // retry_config -- skipped: worker overrides anyway // override_origin -- skipped: will default to tls_cfg.domain diff --git a/packages/core-bridge/src/worker.rs b/packages/core-bridge/src/worker.rs index cefffa08d..9f4150979 100644 --- a/packages/core-bridge/src/worker.rs +++ b/packages/core-bridge/src/worker.rs @@ -541,6 +541,7 @@ mod config { plugins: Vec, workflow_failure_errors: HashSet, workflow_types_to_failure_errors: HashMap>, + disable_payload_error_limit: bool, } #[derive(TryFromJs)] @@ -642,6 +643,7 @@ mod config { .workflow_types_to_failure_errors(into_core_workflow_error_map_of_sets( self.workflow_types_to_failure_errors, )) + .disable_payload_error_limit(self.disable_payload_error_limit) .build() .map_err(|err| BridgeError::TypeError { message: format!("Failed to convert WorkerOptions to CoreWorkerConfig: {err}"), diff --git a/packages/core-bridge/ts/native.ts b/packages/core-bridge/ts/native.ts index 20d4ac7a6..7b2b720b4 100644 --- a/packages/core-bridge/ts/native.ts +++ b/packages/core-bridge/ts/native.ts @@ -140,6 +140,8 @@ export interface ClientOptions { headers: Option>; apiKey: Option; disableErrorCodeMetricTags: boolean; + payloadsWarnSize: number; + memoWarnSize: number; } export interface TlsOptions { @@ -258,6 +260,7 @@ export interface WorkerOptions { plugins: string[]; workflowFailureErrors: WorkflowErrorType[]; workflowTypesToFailureErrors: Record; + disablePayloadErrorLimit: boolean; } export type PollerBehavior = diff --git a/packages/test/src/test-bridge.ts b/packages/test/src/test-bridge.ts index ea391cd92..f50d1efe2 100644 --- a/packages/test/src/test-bridge.ts +++ b/packages/test/src/test-bridge.ts @@ -256,6 +256,8 @@ const GenericConfigs = { headers: null, apiKey: null, disableErrorCodeMetricTags: false, + payloadsWarnSize: 512 * 1024, + memoWarnSize: 2 * 1024, } satisfies native.ClientOptions, }, worker: { @@ -317,6 +319,7 @@ const GenericConfigs = { plugins: [], workflowFailureErrors: [], workflowTypesToFailureErrors: {}, + disablePayloadErrorLimit: false, } satisfies native.WorkerOptions, }, ephemeralServer: { diff --git a/packages/test/src/test-payload-size-limits.ts b/packages/test/src/test-payload-size-limits.ts new file mode 100644 index 000000000..9b9e19870 --- /dev/null +++ b/packages/test/src/test-payload-size-limits.ts @@ -0,0 +1,150 @@ +/** + * Payload/memo size-limit enforcement lives in sdk-core. These tests only assert that the TS + * plumbing reaches core: an oversized worker completion is failed proactively (with a forwarded + * `[TMPRL1103]` error log), the `disablePayloadErrorLimit` opt-out lets the oversized payload reach + * (and be rejected by) the server, and the connection's `payloadsWarnSize` threshold produces a + * forwarded `[TMPRL1103]` warning. + * + * Requires the ephemeral dev server (downloaded by `TestWorkflowEnvironment.createLocal`). + */ +import test from 'ava'; +import { DefaultLogger, NativeConnection, Runtime, makeTelemetryFilterString } from '@temporalio/worker'; +import type { LogEntry } from '@temporalio/worker'; +import { WorkflowFailedError } from '@temporalio/client'; +import { TestWorkflowEnvironment } from '@temporalio/testing'; +import { waitUntil } from '@temporalio/test-helpers'; +import { Worker } from './helpers'; +import * as activities from './activities'; +import { payloadSizeLimitsWorkflow } from './workflows/payload-size-limits'; + +const PAYLOAD_ERROR_LIMIT = 10 * 1024; +const PAYLOAD_LIMITS_EXTRA_ARGS = [ + '--dynamic-config-value', + `limit.blobSize.error=${PAYLOAD_ERROR_LIMIT}`, + // Warn limit must be specified to have the server enforce the error limit. + '--dynamic-config-value', + `limit.blobSize.warn=${2 * 1024}`, +]; + +// Capture core logs forwarded to the TS logger. Installed once for the file's worker process. +const capturedLogs: LogEntry[] = []; +Runtime.install({ + logger: new DefaultLogger('WARN', (entry) => capturedLogs.push(entry)), + telemetryOptions: { + logging: { + // Default filter; temporalio_common must be admitted (it is) for [TMPRL1103] to be visible. + filter: makeTelemetryFilterString({ core: 'WARN', other: 'ERROR' }), + forward: {}, + }, + }, +}); + +function hasLog(level: string, messageSubstring: string): boolean { + return capturedLogs.some((e) => e.level === level && e.message.includes(messageSubstring)); +} + +async function withLocalEnv(fn: (env: TestWorkflowEnvironment) => Promise): Promise { + const env = await TestWorkflowEnvironment.createLocal({ + server: { extraArgs: PAYLOAD_LIMITS_EXTRA_ARGS }, + }); + try { + await fn(env); + } finally { + await env.teardown(); + } +} + +test.serial('oversized worker completion is failed by core with a [TMPRL1103] error log', async (t) => { + capturedLogs.length = 0; + await withLocalEnv(async (env) => { + const worker = await Worker.create({ + connection: env.nativeConnection, + namespace: env.namespace, + taskQueue: 'payload-size-limits', + workflowsPath: require.resolve('./workflows'), + activities, + }); + + // Core repeatedly fails the workflow task (PAYLOADS_TOO_LARGE) for the oversized result, so the + // workflow never completes and hits its execution timeout. + await t.throwsAsync( + worker.runUntil( + env.client.workflow.execute(payloadSizeLimitsWorkflow, { + args: [{ activityInputDataSize: 0, workflowOutputDataSize: PAYLOAD_ERROR_LIMIT + 1024 }], + taskQueue: 'payload-size-limits', + workflowId: `wf-${Date.now()}`, + workflowExecutionTimeout: '5s', + }) + ), + { instanceOf: WorkflowFailedError } + ); + + t.true(hasLog('ERROR', '[TMPRL1103] Attempted to upload payloads with size that exceeded the error limit.')); + }); +}); + +test.serial('disablePayloadErrorLimit sends the oversized payload to the server', async (t) => { + await withLocalEnv(async (env) => { + const worker = await Worker.create({ + connection: env.nativeConnection, + namespace: env.namespace, + taskQueue: 'payload-size-limits-optout', + workflowsPath: require.resolve('./workflows'), + activities, + disablePayloadErrorLimit: 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. + await t.throwsAsync( + worker.runUntil( + env.client.workflow.execute(payloadSizeLimitsWorkflow, { + args: [{ activityInputDataSize: PAYLOAD_ERROR_LIMIT + 1024, workflowOutputDataSize: 0 }], + taskQueue: 'payload-size-limits-optout', + workflowId: `wf-${Date.now()}`, + workflowExecutionTimeout: '10s', + }) + ), + { instanceOf: WorkflowFailedError } + ); + }); +}); + +test.serial('connection payloadsWarnSize produces a forwarded [TMPRL1103] warning', async (t) => { + capturedLogs.length = 0; + await withLocalEnv(async (env) => { + // The warn threshold is configured on the (worker's) connection and forwarded to core. + const connection = await NativeConnection.connect({ + address: env.address, + payloadLimits: { payloadsWarnSize: 1024 }, + }); + try { + const worker = await Worker.create({ + connection, + namespace: env.namespace, + taskQueue: 'payload-size-limits-warn', + workflowsPath: require.resolve('./workflows'), + activities, + }); + + // 2KiB result is above the 1KiB warn threshold but below the 10KiB error limit, so the + // workflow completes and a warning is logged. + await worker.runUntil( + env.client.workflow.execute(payloadSizeLimitsWorkflow, { + args: [{ activityInputDataSize: 0, workflowOutputDataSize: 2 * 1024 }], + taskQueue: 'payload-size-limits-warn', + workflowId: `wf-${Date.now()}`, + workflowExecutionTimeout: '10s', + }) + ); + + // Core forwards logs on a buffered interval, so the warning may not be visible the instant + // `runUntil` resolves for this short-lived workflow; wait for it to arrive. + const warnMessage = '[TMPRL1103] Attempted to upload payloads with size that exceeded the warning limit.'; + await waitUntil(async () => hasLog('WARN', warnMessage), 10_000); + t.true(hasLog('WARN', warnMessage)); + } finally { + await connection.close(); + } + }); +}); diff --git a/packages/test/src/workflows/index.ts b/packages/test/src/workflows/index.ts index d998d067d..7a002f5b6 100644 --- a/packages/test/src/workflows/index.ts +++ b/packages/test/src/workflows/index.ts @@ -86,6 +86,7 @@ export * from './throw-async'; export * from './trailing-timer'; export * from './try-to-continue-after-completion'; export * from './two-strings'; +export * from './payload-size-limits'; // unblockSignal is already defined in ./definitions, don't re-export it. // The reason it is redefined is for completeness of the snippet. export { isBlockedQuery, unblockOrCancel } from './unblock-or-cancel'; diff --git a/packages/test/src/workflows/payload-size-limits.ts b/packages/test/src/workflows/payload-size-limits.ts new file mode 100644 index 000000000..634df98b6 --- /dev/null +++ b/packages/test/src/workflows/payload-size-limits.ts @@ -0,0 +1,23 @@ +/** + * Workflow used by test-payload-size-limits: optionally schedules an activity with a large input, + * then returns a string of the requested size. Used to exercise sdk-core's payload size-limit + * enforcement (proactive task failure / warning logs). + */ +import { proxyActivities } from '@temporalio/workflow'; +import type * as activities from '../activities'; + +const { echo } = proxyActivities({ + scheduleToCloseTimeout: '5s', +}); + +export interface PayloadSizeLimitsInput { + activityInputDataSize: number; + workflowOutputDataSize: number; +} + +export async function payloadSizeLimitsWorkflow(input: PayloadSizeLimitsInput): Promise { + if (input.activityInputDataSize > 0) { + await echo('i'.repeat(input.activityInputDataSize)); + } + return 'o'.repeat(input.workflowOutputDataSize); +} diff --git a/packages/worker/src/connection-options.ts b/packages/worker/src/connection-options.ts index 645264791..735e8b955 100644 --- a/packages/worker/src/connection-options.ts +++ b/packages/worker/src/connection-options.ts @@ -44,6 +44,29 @@ export interface NoneGrpcCompressionConfig { */ export type GrpcCompressionConfig = GzipGrpcCompressionConfig | NoneGrpcCompressionConfig; +/** + * Payload size-limit configuration for a connection. + * + * @experimental Payload size-limit enforcement is an experimental feature; APIs may change without notice. + */ +export interface PayloadLimitsConfig { + /** + * Warning threshold, in bytes, for the size of an outbound payload-bearing field. Over-threshold + * fields are logged but still sent to the server. Set to `0` to disable. + * + * @default 512KiB + */ + payloadsWarnSize?: number; + + /** + * Warning threshold, in bytes, for outbound memo size. Over-threshold memos are logged but still + * sent to the server. Set to `0` to disable. + * + * @default 2KiB + */ + memoWarnSize?: number; +} + /** * The default Temporal Server's TCP port for public gRPC connections. */ @@ -112,6 +135,14 @@ export interface NativeConnectionOptions { */ disableErrorCodeMetricTags?: boolean; + /** + * Payload size-limit options for this connection. If unset, defaults of 512KiB (payloads) and + * 2KiB (memo) are used. + * + * @experimental Payload size-limit enforcement is an experimental feature; APIs may change without notice. + */ + payloadLimits?: PayloadLimitsConfig; + /** * List of plugins to register with the native connection. * @@ -203,5 +234,7 @@ export function toNativeClientOptions(options: NativeConnectionOptions): native. headers, apiKey: options.apiKey ?? null, disableErrorCodeMetricTags: options.disableErrorCodeMetricTags ?? false, + payloadsWarnSize: options.payloadLimits?.payloadsWarnSize ?? 512 * 1024, + memoWarnSize: options.payloadLimits?.memoWarnSize ?? 2 * 1024, }; } diff --git a/packages/worker/src/worker-options.ts b/packages/worker/src/worker-options.ts index 84c604cd3..80a0114c0 100644 --- a/packages/worker/src/worker-options.ts +++ b/packages/worker/src/worker-options.ts @@ -225,6 +225,15 @@ export interface WorkerOptions { */ enableNonLocalActivities?: boolean; + /** + * If set to `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). + * + * @experimental Payload size-limit enforcement is an experimental feature; APIs may change without notice. + */ + disablePayloadErrorLimit?: boolean; + /** * Limits the number of Activities per second that this Worker will process. (Does not limit the number of Local * Activities.) The Worker will not poll for new Activities if by doing so it might receive and execute an Activity @@ -1181,6 +1190,7 @@ export function toNativeWorkerOptions(opts: CompiledWorkerOptionsWithBuildId): n plugins: opts.plugins?.map((p) => p.name) ?? [], workflowFailureErrors, workflowTypesToFailureErrors, + disablePayloadErrorLimit: opts.disablePayloadErrorLimit ?? false, }; }