💥 Enforce payload and memo size limits

CHANGELOG.md

@@ -23,6 +23,10 @@ to docs, or any other relevant information.
 - `WorkflowOutboundCallsInterceptor` has new `sleep(Duration, TimerOptions)` and
   `await(Duration, TimerOptions, String, Supplier)` methods. Implementations that don't extend
   `WorkflowOutboundCallsInterceptorBase` need to add them, and Base subclasses override them to see calls with options.
+- Workers now retryably fail a workflow or activity task whose payloads exceed the namespace's size
+  limit, instead of sending it to the server. Disable with
+  `WorkerOptions.Builder.setDisablePayloadErrorLimit`; tune warnings with the experimental
+  `WorkflowServiceStubsOptions.Builder.setPayloadLimits`.
 
 ### Added
 - Added experimental `ChildWorkflowOptions.Builder.setVersioningOverride` and

temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextFactoryImpl.java

@@ -4,6 +4,7 @@
 import io.temporal.client.WorkflowClient;
 import io.temporal.common.converter.DataConverter;
 import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
+import io.temporal.internal.payload.limits.PayloadErrorLimits;
 import io.temporal.internal.payload.storage.ExternalStorageRunner;
 import java.nio.ByteBuffer;
 import java.time.Duration;
@@ -12,6 +13,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.function.Supplier;
 import javax.annotation.Nullable;
 
 public class ActivityExecutionContextFactoryImpl implements ActivityExecutionContextFactory {
@@ -24,9 +26,14 @@ public class ActivityExecutionContextFactoryImpl implements ActivityExecutionCon
   private final ScheduledExecutorService heartbeatExecutor;
   private final ManualActivityCompletionClientFactory manualCompletionClientFactory;
   private final @Nullable ExternalStorageRunner externalStorage;
+  private final Supplier<PayloadErrorLimits> payloadErrorLimits;
   private final ConcurrentMap<ByteBuffer, ActivityExecutionContextImpl> activeContexts =
       new ConcurrentHashMap<>();
 
+  /**
+   * @param payloadErrorLimits read when each activity starts, because the namespace's limits are
+   *     only known once the worker factory has started.
+   */
   public ActivityExecutionContextFactoryImpl(
       WorkflowClient client,
       String identity,
@@ -35,7 +42,9 @@ public ActivityExecutionContextFactoryImpl(
       Duration defaultHeartbeatThrottleInterval,
       DataConverter dataConverter,
       ScheduledExecutorService heartbeatExecutor,
-      @Nullable ExternalStorageRunner externalStorage) {
+      @Nullable ExternalStorageRunner externalStorage,
+      Supplier<PayloadErrorLimits> payloadErrorLimits) {
+    this.payloadErrorLimits = Objects.requireNonNull(payloadErrorLimits);
     this.client = Objects.requireNonNull(client);
     this.identity = identity;
     this.namespace = Objects.requireNonNull(namespace);
@@ -69,7 +78,8 @@ public InternalActivityExecutionContext createContext(
             maxHeartbeatThrottleInterval,
             defaultHeartbeatThrottleInterval,
             () -> cleanupContext(info.getTaskToken(), false),
-            externalStorage);
+            externalStorage,
+            payloadErrorLimits.get());
     activeContexts.put(taskToken, context);
     return context;
   }

temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextImpl.java

@@ -10,6 +10,8 @@
 import io.temporal.common.CancellationToken;
 import io.temporal.common.converter.DataConverter;
 import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
+import io.temporal.internal.payload.limits.PayloadErrorLimits;
+import io.temporal.internal.payload.limits.PayloadLimitViolation;
 import io.temporal.internal.payload.storage.ExternalStorageRunner;
 import io.temporal.payload.context.ActivitySerializationContext;
 import io.temporal.workflow.Functions;
@@ -58,7 +60,8 @@ class ActivityExecutionContextImpl implements InternalActivityExecutionContext {
       Duration maxHeartbeatThrottleInterval,
       Duration defaultHeartbeatThrottleInterval,
       Functions.Proc closeCallback,
-      @Nullable ExternalStorageRunner externalStorage) {
+      @Nullable ExternalStorageRunner externalStorage,
+      PayloadErrorLimits payloadErrorLimits) {
     this.client = client;
     this.activity = activity;
     this.metricsScope = metricsScope;
@@ -77,7 +80,8 @@ class ActivityExecutionContextImpl implements InternalActivityExecutionContext {
             identity,
             maxHeartbeatThrottleInterval,
             defaultHeartbeatThrottleInterval,
-            externalStorage);
+            externalStorage,
+            payloadErrorLimits);
   }
 
   /**
@@ -88,6 +92,17 @@ public <V> void heartbeat(V details) throws ActivityCompletionException {
     heartbeatContext.heartbeat(details);
   }
 
+  @Override
+  public boolean isTaskReported() {
+    return heartbeatContext.isTaskReported();
+  }
+
+  @Override
+  @Nullable
+  public PayloadLimitViolation getPayloadLimitViolation() {
+    return heartbeatContext.getPayloadLimitViolation();
+  }
+
   @Override
   public <V> Optional<V> getHeartbeatDetails(Class<V> detailsClass) {
     return getHeartbeatDetails(detailsClass, detailsClass);

temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskExecutors.java

@@ -8,6 +8,7 @@
 import io.temporal.api.common.v1.Payload;
 import io.temporal.api.common.v1.Payloads;
 import io.temporal.api.workflowservice.v1.RespondActivityTaskCompletedRequest;
+import io.temporal.api.workflowservice.v1.RespondActivityTaskFailedRequest;
 import io.temporal.client.ActivityCanceledException;
 import io.temporal.common.context.ContextPropagator;
 import io.temporal.common.converter.DataConverter;
@@ -16,7 +17,10 @@
 import io.temporal.common.interceptors.ActivityInboundCallsInterceptor.ActivityOutput;
 import io.temporal.common.interceptors.Header;
 import io.temporal.common.interceptors.WorkerInterceptor;
+import io.temporal.failure.ApplicationFailure;
 import io.temporal.internal.common.FailureUtils;
+import io.temporal.internal.common.PayloadsTooLargeFailure;
+import io.temporal.internal.payload.limits.PayloadLimitViolation;
 import io.temporal.internal.worker.ActivityTaskHandler;
 import io.temporal.payload.context.ActivitySerializationContext;
 import io.temporal.serviceclient.CheckedExceptionWrapper;
@@ -91,18 +95,27 @@ public ActivityTaskHandler.Result execute(ActivityInfoInternal info, Scope metri
           return new ActivityTaskHandler.Result(
               info.getActivityId(), null, null, null, context.isUseLocalManualCompletion());
         }
+        ActivityTaskHandler.Result heartbeatOutcome =
+            heartbeatOutcome(context, info, dataConverterWithActivityContext);
+        if (heartbeatOutcome != null) {
+          return heartbeatOutcome;
+        }
 
         return this.constructSuccessfulResultValue(info, result, dataConverterWithActivityContext);
       } catch (Throwable e) {
         Throwable ex = CheckedExceptionWrapper.unwrap(e);
         boolean local = info.isLocal();
         if (ex instanceof ActivityCanceledException) {
-          log.info(
-              "{} canceled. ActivityId={}, activityType={}, attempt={}",
-              local ? "Local activity" : "Activity",
-              info.getActivityId(),
-              info.getActivityType(),
-              info.getAttempt());
+          // An oversized heartbeat cancels the activity only to fail it, and the payload limit
+          // validator already logged why.
+          if (context.getPayloadLimitViolation() == null) {
+            log.info(
+                "{} canceled. ActivityId={}, activityType={}, attempt={}",
+                local ? "Local activity" : "Activity",
+                info.getActivityId(),
+                info.getActivityType(),
+                info.getAttempt());
+          }
         } else if (FailureUtils.isBenignApplicationFailure(ex)) {
           log.debug(
               "{} failure. ActivityId={}, activityType={}, attempt={}",
@@ -121,6 +134,11 @@ public ActivityTaskHandler.Result execute(ActivityInfoInternal info, Scope metri
               ex);
         }
 
+        ActivityTaskHandler.Result heartbeatOutcome =
+            heartbeatOutcome(context, info, dataConverterWithActivityContext);
+        if (heartbeatOutcome != null) {
+          return heartbeatOutcome;
+        }
         return mapToActivityFailure(
             ex,
             info.getActivityId(),
@@ -142,6 +160,36 @@ public ActivityTaskHandler.Result execute(ActivityInfoInternal info, Scope metri
       }
     }
 
+    /**
+     * Returns the result to report when an oversized heartbeat decided the task's outcome, or null.
+     * The result sends nothing if the heartbeat already reported the failure, and otherwise reports
+     * that failure in place of whatever the activity did.
+     */
+    @Nullable
+    private static ActivityTaskHandler.Result heartbeatOutcome(
+        InternalActivityExecutionContext context,
+        ActivityInfoInternal info,
+        DataConverter dataConverterWithActivityContext) {
+      if (context.isTaskReported()) {
+        return new ActivityTaskHandler.Result(info.getActivityId(), null, null, null, false);
+      }
+      PayloadLimitViolation violation = context.getPayloadLimitViolation();
+      if (violation == null) {
+        return null;
+      }
+      ApplicationFailure failure = PayloadsTooLargeFailure.of(violation);
+      RespondActivityTaskFailedRequest request =
+          RespondActivityTaskFailedRequest.newBuilder()
+              .setFailure(dataConverterWithActivityContext.exceptionToFailure(failure))
+              .build();
+      return new ActivityTaskHandler.Result(
+          info.getActivityId(),
+          null,
+          new ActivityTaskHandler.Result.TaskFailedResult(request, failure),
+          null,
+          false);
+    }
+
     abstract ActivityInboundCallsInterceptor createRootInboundInterceptor();
 
     abstract Object getActivity();

temporal-sdk/src/main/java/io/temporal/internal/activity/HeartbeatContext.java

@@ -3,8 +3,10 @@
 import io.temporal.client.ActivityCanceledException;
 import io.temporal.client.ActivityCompletionException;
 import io.temporal.common.CancellationToken;
+import io.temporal.internal.payload.limits.PayloadLimitViolation;
 import java.lang.reflect.Type;
 import java.util.Optional;
+import javax.annotation.Nullable;
 
 interface HeartbeatContext {
 
@@ -35,4 +37,17 @@ interface HeartbeatContext {
 
   /** Cancel any pending heartbeat and discard cached heartbeat details. */
   void cancelOutstandingHeartbeat();
+
+  /**
+   * Returns whether this context already reported the activity task's outcome to the server, so the
+   * activity's own result must not be sent.
+   */
+  boolean isTaskReported();
+
+  /**
+   * Returns the payload limit violation that failed a heartbeat, or null. Unless this context
+   * already reported it, the activity must report the violation instead of its own outcome.
+   */
+  @Nullable
+  PayloadLimitViolation getPayloadLimitViolation();
 }