💥 feature: payloads/memo size enforcement

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

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<HashMap<String, MetadataValue>>,
         api_key: Option<String>,
         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

packages/core-bridge/src/worker.rs

@@ -541,6 +541,7 @@ mod config {
         plugins: Vec<String>,
         workflow_failure_errors: HashSet<WorkflowErrorType>,
         workflow_types_to_failure_errors: HashMap<String, HashSet<WorkflowErrorType>>,
+        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}"),

packages/core-bridge/ts/native.ts

@@ -140,6 +140,8 @@ export interface ClientOptions {
   headers: Option<Record<string, MetadataValue>>;
   apiKey: Option<string>;
   disableErrorCodeMetricTags: boolean;
+  payloadsWarnSize: number;
+  memoWarnSize: number;
 }
 
 export interface TlsOptions {
@@ -258,6 +260,7 @@ export interface WorkerOptions {
   plugins: string[];
   workflowFailureErrors: WorkflowErrorType[];
   workflowTypesToFailureErrors: Record<string, WorkflowErrorType[]>;
+  disablePayloadErrorLimit: boolean;
 }
 
 export type PollerBehavior =

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: {

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<void>): Promise<void> {
+  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();
+    }
+  });
+});

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';

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<typeof activities>({
+  scheduleToCloseTimeout: '5s',
+});
+
+export interface PayloadSizeLimitsInput {
+  activityInputDataSize: number;
+  workflowOutputDataSize: number;
+}
+
+export async function payloadSizeLimitsWorkflow(input: PayloadSizeLimitsInput): Promise<string> {
+  if (input.activityInputDataSize > 0) {
+    await echo('i'.repeat(input.activityInputDataSize));
+  }
+  return 'o'.repeat(input.workflowOutputDataSize);
+}

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,
   };
 }

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,
   };
 }