From c9ab565196a013309bfd3f5b9130c1f5bfec2886 Mon Sep 17 00:00:00 2001 From: jmaeagle99 <44687433+jmaeagle99@users.noreply.github.com> Date: Fri, 9 Oct 2026 20:04:15 -0700 Subject: [PATCH] =?UTF-8?q?=F0=9F=92=A5=20Enforce=20payload=20and=20memo?= =?UTF-8?q?=20size=20limits?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 4 + .../ActivityExecutionContextFactoryImpl.java | 14 +- .../ActivityExecutionContextImpl.java | 19 +- .../activity/ActivityTaskExecutors.java | 58 ++- .../internal/activity/HeartbeatContext.java | 15 + .../activity/HeartbeatContextImpl.java | 111 ++++- .../InternalActivityExecutionContext.java | 16 + .../LocalActivityExecutionContextImpl.java | 13 + .../internal/client/ActivityClientHelper.java | 10 + .../common/PayloadsTooLargeFailure.java | 38 ++ .../internal/worker/ActivityWorker.java | 27 ++ .../worker/NamespaceCapabilities.java | 20 + .../temporal/internal/worker/NexusWorker.java | 50 +++ .../internal/worker/SingleWorkerOptions.java | 22 + .../internal/worker/SyncActivityWorker.java | 3 +- .../internal/worker/WorkflowWorker.java | 56 +++ .../main/java/io/temporal/worker/Worker.java | 1 + .../io/temporal/worker/WorkerOptions.java | 42 ++ .../ActivityExecutionContextImplTest.java | 4 +- .../ActivityTaskHandlerPayloadLimitsTest.java | 223 +++++++++ .../activity/HeartbeatContextImplTest.java | 210 ++++++++- .../ActivityWorkerPayloadLimitsTest.java | 219 +++++++++ .../internal/worker/NexusWorkerTest.java | 150 ++++++- .../PayloadErrorLimitsWorkerOptionsTest.java | 56 +++ .../internal/worker/WorkflowWorkerTest.java | 199 ++++++++ .../io/temporal/worker/WorkerOptionsTest.java | 14 + .../workflow/GrpcMessageTooLargeTest.java | 12 +- .../workflow/PayloadLimitsDisabledTest.java | 40 ++ .../temporal/workflow/PayloadLimitsTest.java | 208 +++++++++ temporal-serviceclient/build.gradle | 2 +- .../internal/payload/limits/LimitClass.java | 2 +- .../payload/limits/LimitSeverity.java | 2 +- .../payload/limits/PayloadErrorLimits.java | 38 ++ .../payload/limits/PayloadLimitViolation.java | 16 +- .../PayloadLimitViolationException.java | 40 ++ .../limits/PayloadLimitsInterceptor.java | 183 ++++++++ .../serviceclient/ChannelManager.java | 14 + .../io/temporal/serviceclient/MetricsTag.java | 1 + .../serviceclient/PayloadLimitsOptions.java | 122 +++++ .../WorkflowServiceStubsOptions.java | 53 ++- .../PayloadLimitsInterceptorTest.java | 424 ++++++++++++++++++ .../TestActivityEnvironmentInternal.java | 4 +- 42 files changed, 2715 insertions(+), 40 deletions(-) create mode 100644 temporal-sdk/src/main/java/io/temporal/internal/common/PayloadsTooLargeFailure.java create mode 100644 temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityTaskHandlerPayloadLimitsTest.java create mode 100644 temporal-sdk/src/test/java/io/temporal/internal/worker/ActivityWorkerPayloadLimitsTest.java create mode 100644 temporal-sdk/src/test/java/io/temporal/internal/worker/PayloadErrorLimitsWorkerOptionsTest.java create mode 100644 temporal-sdk/src/test/java/io/temporal/workflow/PayloadLimitsDisabledTest.java create mode 100644 temporal-sdk/src/test/java/io/temporal/workflow/PayloadLimitsTest.java create mode 100644 temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadErrorLimits.java create mode 100644 temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitViolationException.java create mode 100644 temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitsInterceptor.java create mode 100644 temporal-serviceclient/src/main/java/io/temporal/serviceclient/PayloadLimitsOptions.java create mode 100644 temporal-serviceclient/src/test/java/io/temporal/serviceclient/PayloadLimitsInterceptorTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 9216848cc4..0299a90ede 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextFactoryImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextFactoryImpl.java index 5f9584180f..2eae8ff5d5 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextFactoryImpl.java +++ b/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; private final ConcurrentMap 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) { + 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; } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextImpl.java index 64cdd75383..b3681d83d3 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextImpl.java +++ b/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 void heartbeat(V details) throws ActivityCompletionException { heartbeatContext.heartbeat(details); } + @Override + public boolean awaitTaskReported() { + return heartbeatContext.awaitTaskReported(); + } + + @Override + @Nullable + public PayloadLimitViolation getPayloadLimitViolation() { + return heartbeatContext.getPayloadLimitViolation(); + } + @Override public Optional getHeartbeatDetails(Class detailsClass) { return getHeartbeatDetails(detailsClass, detailsClass); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskExecutors.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskExecutors.java index 7789bd2716..e308435fdf 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskExecutors.java +++ b/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,34 @@ public ActivityTaskHandler.Result execute(ActivityInfoInternal info, Scope metri } } + /** + * Returns the result to report when an oversized heartbeat decided the task's outcome, or null. + */ + @Nullable + private static ActivityTaskHandler.Result heartbeatOutcome( + InternalActivityExecutionContext context, + ActivityInfoInternal info, + DataConverter dataConverterWithActivityContext) { + if (context.awaitTaskReported()) { + 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(); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/HeartbeatContext.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/HeartbeatContext.java index faef0d7950..3e74c558e7 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/HeartbeatContext.java +++ b/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. Waits for a report that is still being sent. + */ + boolean awaitTaskReported(); + + /** + * 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(); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/HeartbeatContextImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/HeartbeatContextImpl.java index c41f3740d9..c6fdc543fe 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/HeartbeatContextImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/activity/HeartbeatContextImpl.java @@ -1,5 +1,7 @@ package io.temporal.internal.activity; +import static io.temporal.serviceclient.MetricsTag.METRICS_TAGS_CALL_OPTIONS_KEY; + import com.google.common.base.Strings; import com.google.common.base.Throwables; import com.google.protobuf.ByteString; @@ -12,17 +14,25 @@ import io.temporal.api.enums.v1.TimeoutType; import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatRequest; import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatResponse; +import io.temporal.api.workflowservice.v1.RespondActivityTaskFailedRequest; import io.temporal.client.*; import io.temporal.common.CancellationToken; import io.temporal.common.converter.DataConverter; import io.temporal.failure.TimeoutFailure; import io.temporal.internal.client.ActivityClientHelper; +import io.temporal.internal.common.PayloadsTooLargeFailure; import io.temporal.internal.concurrent.structured.CancelSource; +import io.temporal.internal.payload.limits.PayloadErrorLimits; +import io.temporal.internal.payload.limits.PayloadLimitViolation; +import io.temporal.internal.payload.limits.PayloadLimitViolationException; import io.temporal.internal.payload.storage.ActivityStorageTargets; import io.temporal.internal.payload.storage.ExternalStorageRunner; +import io.temporal.internal.retryer.GrpcRetryer; import io.temporal.payload.context.ActivitySerializationContext; import io.temporal.payload.storage.StorageDriverTargetInfo; import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.serviceclient.rpcretry.DefaultStubServiceOperationRpcRetryOptions; +import io.temporal.worker.MetricsType; import java.lang.reflect.Type; import java.time.Duration; import java.util.Optional; @@ -35,6 +45,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import javax.annotation.Nullable; @@ -84,6 +95,14 @@ static long getLocalHeartbeatTimeoutBufferMillis() { private final Optional prevAttemptHeartbeatDetails; private final long heartbeatTimeoutMillis; private final long localHeartbeatTimeoutBufferMillis; + private final PayloadErrorLimits payloadErrorLimits; + private final GrpcRetryer grpcRetryer; + private volatile @Nullable PayloadLimitViolation payloadLimitViolation; + private final AtomicReference pendingPayloadLimitReport = + new AtomicReference<>(); + // Completes once that failure has been sent or given up on. + private volatile @Nullable CompletableFuture payloadLimitReport; + private volatile boolean taskReported; // turned into true on a reception of the first heartbeat private boolean receivedAHeartbeat = false; @@ -114,7 +133,8 @@ public HeartbeatContextImpl( String identity, Duration maxHeartbeatThrottleInterval, Duration defaultHeartbeatThrottleInterval, - @Nullable ExternalStorageRunner externalStorage) { + @Nullable ExternalStorageRunner externalStorage, + PayloadErrorLimits payloadErrorLimits) { this( service, namespace, @@ -126,6 +146,7 @@ public HeartbeatContextImpl( maxHeartbeatThrottleInterval, defaultHeartbeatThrottleInterval, externalStorage, + payloadErrorLimits, getLocalHeartbeatTimeoutBufferMillis()); } @@ -140,8 +161,11 @@ public HeartbeatContextImpl( Duration maxHeartbeatThrottleInterval, Duration defaultHeartbeatThrottleInterval, @Nullable ExternalStorageRunner externalStorage, + PayloadErrorLimits payloadErrorLimits, long localHeartbeatTimeoutBufferMillis) { this.service = service; + this.payloadErrorLimits = payloadErrorLimits; + this.grpcRetryer = new GrpcRetryer(service.getServerCapabilities()); this.metricsScope = metricsScope; this.dataConverter = dataConverter; this.externalStorage = externalStorage; @@ -200,6 +224,7 @@ public void heartbeat(V details) throws ActivityCompletionException { cancellationSource.token().throwIfCancellationRequested(); } finally { lock.unlock(); + reportPendingPayloadLimitFailure(); } } @@ -299,7 +324,17 @@ public CancellationToken getCancellationToken() { private void doHeartBeatLocked(Object details) { long nextHeartbeatDelay; try { - sendHeartbeatRequest(details); + // An activity failed for an oversized heartbeat sends no more heartbeats: each would fail + // the activity again. The violation is checked again after the send, because the send is + // what finds the first one. + if (payloadLimitViolation == null) { + sendHeartbeatRequest(details); + } + if (payloadLimitViolation != null) { + hasOutstandingHeartbeat = false; + scheduledHeartbeat = null; + return; + } hasOutstandingHeartbeat = false; nextHeartbeatDelay = heartbeatIntervalMillis; // Reset the local heartbeat timeout deadline only on successful send. @@ -340,6 +375,7 @@ private void scheduleNextHeartbeatLocked(long delay) { } } finally { lock.unlock(); + reportPendingPayloadLimitFailure(); } }, delay, @@ -457,7 +493,8 @@ private void sendHeartbeatRequest(Object details) { } RecordActivityTaskHeartbeatRequest request = builder.build(); RecordActivityTaskHeartbeatResponse status = - ActivityClientHelper.sendHeartbeatRequest(service, request, metricsScope); + ActivityClientHelper.sendHeartbeatRequest( + service, request, metricsScope, payloadErrorLimits); if (status.getCancelRequested()) { requestCancelLocked(); } else if (status.getActivityReset()) { @@ -468,7 +505,12 @@ private void sendHeartbeatRequest(Object details) { lastException = null; } } catch (StatusRuntimeException e) { - if (e.getStatus().getCode() == Status.Code.NOT_FOUND) { + Optional violation = PayloadLimitViolationException.find(e); + if (violation.isPresent()) { + failActivityWithPayloadsTooLarge(violation.get()); + // Cancellation is how the running activity learns that it was failed. + requestCancelLocked(); + } else if (e.getStatus().getCode() == Status.Code.NOT_FOUND) { lastException = new ActivityNotExistsException(info, e); } else if (e.getStatus().getCode() == Status.Code.INVALID_ARGUMENT || e.getStatus().getCode() == Status.Code.FAILED_PRECONDITION) { @@ -479,6 +521,67 @@ private void sendHeartbeatRequest(Object details) { } } + /** + * Queues the failure for a heartbeat over the payload error limits. {@link + * #reportPendingPayloadLimitFailure} sends it once the lock is released, so that retrying the + * call does not hold the lock. + */ + private void failActivityWithPayloadsTooLarge(PayloadLimitViolation violation) { + PayloadsTooLargeFailure.count(metricsScope, MetricsType.ACTIVITY_EXEC_FAILED_COUNTER); + // These are set before cancellation is requested, so an activity that sees the cancellation + // also sees how its outcome must be reported. + payloadLimitReport = new CompletableFuture<>(); + pendingPayloadLimitReport.set( + RespondActivityTaskFailedRequest.newBuilder() + .setTaskToken(ByteString.copyFrom(info.getTaskToken())) + .setNamespace(namespace) + .setIdentity(identity) + .setFailure( + dataConverterWithActivityContext.exceptionToFailure( + PayloadsTooLargeFailure.of(violation))) + .build()); + payloadLimitViolation = violation; + } + + /** Sends the failure queued by an oversized heartbeat, if any. Called without the lock held. */ + private void reportPendingPayloadLimitFailure() { + RespondActivityTaskFailedRequest request = pendingPayloadLimitReport.getAndSet(null); + if (request == null) { + return; + } + try { + grpcRetryer.retry( + () -> + service + .blockingStub() + .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) + .respondActivityTaskFailed(request), + new GrpcRetryer.GrpcRetryerOptions( + DefaultStubServiceOperationRpcRetryOptions.INSTANCE, null)); + taskReported = true; + } catch (Exception e) { + // The activity reports this failure itself when it finishes, instead of its own outcome. + log.warn("Failed to fail activity after an oversized heartbeat", e); + } finally { + payloadLimitReport.complete(null); + } + } + + @Override + public boolean awaitTaskReported() { + CompletableFuture report = payloadLimitReport; + if (report != null) { + report.join(); + } + return taskReported; + } + + @Override + @Nullable + public PayloadLimitViolation getPayloadLimitViolation() { + return payloadLimitViolation; + } + private void requestCancelLocked() { ActivityCanceledException exception = new ActivityCanceledException(info); lastException = exception; diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/InternalActivityExecutionContext.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/InternalActivityExecutionContext.java index 8b6fbe997e..51c66dc99f 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/InternalActivityExecutionContext.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/activity/InternalActivityExecutionContext.java @@ -1,6 +1,8 @@ package io.temporal.internal.activity; import io.temporal.activity.ActivityExecutionContext; +import io.temporal.internal.payload.limits.PayloadLimitViolation; +import javax.annotation.Nullable; /** * Internal context object passed to an Activity implementation, providing more internal details @@ -15,4 +17,18 @@ public interface InternalActivityExecutionContext extends ActivityExecutionConte /** Cancel any pending heartbeat and discard cached heartbeat details. */ void cancelOutstandingHeartbeat(); + + /** + * Returns whether the activity task's outcome was already reported to the server, for example + * because an oversized heartbeat failed it, so the activity's own result must not be sent. Waits + * for a report that is still being sent. + */ + boolean awaitTaskReported(); + + /** + * Returns the payload limit violation that failed a heartbeat, or null. Unless the task was + * already reported, the activity must report the violation instead of its own outcome. + */ + @Nullable + PayloadLimitViolation getPayloadLimitViolation(); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/LocalActivityExecutionContextImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/LocalActivityExecutionContextImpl.java index 0e8282c580..58f0b5ad72 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/LocalActivityExecutionContextImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/activity/LocalActivityExecutionContextImpl.java @@ -7,8 +7,10 @@ import io.temporal.client.ActivityCompletionException; import io.temporal.client.WorkflowClient; 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; class LocalActivityExecutionContextImpl implements InternalActivityExecutionContext { private final WorkflowClient client; @@ -115,4 +117,15 @@ public WorkflowClient getWorkflowClient() { public Object getInstance() { return activity; } + + @Override + public boolean awaitTaskReported() { + return false; + } + + @Override + @Nullable + public PayloadLimitViolation getPayloadLimitViolation() { + return null; + } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityClientHelper.java b/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityClientHelper.java index cedbf38684..e70eb4361b 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityClientHelper.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityClientHelper.java @@ -8,6 +8,7 @@ import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatByIdResponse; import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatRequest; import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatResponse; +import io.temporal.internal.payload.limits.PayloadErrorLimits; import io.temporal.serviceclient.WorkflowServiceStubs; /** @@ -22,9 +23,18 @@ public static RecordActivityTaskHeartbeatResponse sendHeartbeatRequest( WorkflowServiceStubs service, RecordActivityTaskHeartbeatRequest request, Scope metricsScope) { + return sendHeartbeatRequest(service, request, metricsScope, PayloadErrorLimits.NONE); + } + + public static RecordActivityTaskHeartbeatResponse sendHeartbeatRequest( + WorkflowServiceStubs service, + RecordActivityTaskHeartbeatRequest request, + Scope metricsScope, + PayloadErrorLimits payloadErrorLimits) { return service .blockingStub() .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) + .withOption(PayloadErrorLimits.CALL_OPTIONS_KEY, payloadErrorLimits) .recordActivityTaskHeartbeat(request); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/common/PayloadsTooLargeFailure.java b/temporal-sdk/src/main/java/io/temporal/internal/common/PayloadsTooLargeFailure.java new file mode 100644 index 0000000000..9c57eee667 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/common/PayloadsTooLargeFailure.java @@ -0,0 +1,38 @@ +package io.temporal.internal.common; + +import com.uber.m3.tally.Scope; +import io.temporal.failure.ApplicationFailure; +import io.temporal.internal.payload.limits.PayloadLimitViolation; +import io.temporal.internal.payload.limits.PayloadLimitViolationException; +import io.temporal.serviceclient.MetricsTag; +import java.util.Collections; + +/** + * Builds the failure a worker reports instead of a completion that exceeds the namespace's payload + * error limits. The failure is retryable, so that a corrected workflow or activity can be + * redeployed and continue. + */ +public final class PayloadsTooLargeFailure { + private PayloadsTooLargeFailure() {} + + public static ApplicationFailure of(PayloadLimitViolation violation) { + ApplicationFailure failure = + ApplicationFailure.newBuilder() + .setMessage(violation.getMessage()) + .setType(PayloadLimitViolationException.FAILURE_TYPE) + .build(); + // The worker's stack trace says nothing about the oversized payload, so it is not sent. + failure.setStackTrace(new StackTraceElement[0]); + return failure; + } + + /** Counts one task failed for exceeding the payload error limits on {@code counter}. */ + public static void count(Scope metricsScope, String counter) { + metricsScope + .tagged( + Collections.singletonMap( + MetricsTag.TASK_FAILURE_TYPE, MetricsTag.TASK_FAILURE_VALUE_PAYLOADS_TOO_LARGE)) + .counter(counter) + .inc(1); + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java index 5aa5293175..d8b95ab155 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java @@ -14,10 +14,14 @@ import io.temporal.api.workflowservice.v1.*; import io.temporal.failure.ApplicationFailure; import io.temporal.internal.activity.ActivityPollResponseToInfo; +import io.temporal.internal.common.PayloadsTooLargeFailure; import io.temporal.internal.common.ProtobufTimeUtils; import io.temporal.internal.concurrent.structured.CancelSource; import io.temporal.internal.logging.LoggerTag; import io.temporal.internal.logging.PrefixedMdc; +import io.temporal.internal.payload.limits.PayloadErrorLimits; +import io.temporal.internal.payload.limits.PayloadLimitViolation; +import io.temporal.internal.payload.limits.PayloadLimitViolationException; import io.temporal.internal.payload.storage.ActivityStorageTargets; import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.internal.retryer.GrpcRetryer; @@ -408,6 +412,11 @@ private ActivityTaskHandler.Result handleActivity(ActivityTask task, Scope metri sendStorageFailure(taskToken, metricsScope, e.getCause()); return result; } catch (Exception e) { + Optional violation = PayloadLimitViolationException.find(e); + if (violation.isPresent()) { + sendPayloadsTooLargeFailure(taskToken, metricsScope, violation.get()); + return result; + } logExceptionDuringResultReporting(e, pollResponse, result); // TODO this class doesn't report activity success and failure metrics now, instead it's // located inside an activity handler. We should lift it up to this level, @@ -464,6 +473,9 @@ private void sendReply( service .blockingStub() .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) + .withOption( + PayloadErrorLimits.CALL_OPTIONS_KEY, + options.payloadErrorLimits(namespaceCapabilities)) .respondActivityTaskCompleted(request), replyGrpcRetryerOptions); } else { @@ -502,6 +514,9 @@ private void sendReply( service .blockingStub() .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) + .withOption( + PayloadErrorLimits.CALL_OPTIONS_KEY, + options.payloadErrorLimits(namespaceCapabilities)) .respondActivityTaskCanceled(request), replyGrpcRetryerOptions); } @@ -550,6 +565,18 @@ private RespondActivityTaskFailedRequest sendStorageFailure( .setType(ExternalStorageTaskFailure.class.getSimpleName()) .build(); applicationFailure.setStackTrace(new StackTraceElement[0]); + return sendFailure(taskToken, metricsScope, applicationFailure); + } + + private void sendPayloadsTooLargeFailure( + ByteString taskToken, Scope metricsScope, PayloadLimitViolation violation) { + PayloadsTooLargeFailure.count(metricsScope, MetricsType.ACTIVITY_EXEC_FAILED_COUNTER); + sendFailure(taskToken, metricsScope, PayloadsTooLargeFailure.of(violation)); + } + + @SuppressWarnings("deprecation") + private RespondActivityTaskFailedRequest sendFailure( + ByteString taskToken, Scope metricsScope, ApplicationFailure applicationFailure) { RespondActivityTaskFailedRequest request = RespondActivityTaskFailedRequest.newBuilder() .setTaskToken(taskToken) diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/NamespaceCapabilities.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/NamespaceCapabilities.java index e9866089dc..d0b6cda156 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/NamespaceCapabilities.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/NamespaceCapabilities.java @@ -2,8 +2,10 @@ import io.temporal.api.namespace.v1.NamespaceInfo.Capabilities; import io.temporal.api.namespace.v1.NamespaceInfo.Limits; +import io.temporal.internal.payload.limits.PayloadErrorLimits; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; /** * Holds namespace-level capabilities discovered from the server's DescribeNamespace response. A @@ -18,6 +20,8 @@ public final class NamespaceCapabilities { private final AtomicBoolean workerCommands = new AtomicBoolean(false); private final AtomicBoolean workflowTaskCompletionPagination = new AtomicBoolean(false); private final AtomicLong workflowTaskCompletionSizeLimit = new AtomicLong(0); + private final AtomicReference payloadErrorLimits = + new AtomicReference<>(PayloadErrorLimits.NONE); public void setFromCapabilities(Capabilities capabilities) { if (capabilities.getPollerAutoscalingAutoEnroll()) { @@ -42,6 +46,7 @@ public void setFromCapabilities(Capabilities capabilities) { public void setFromLimits(Limits limits) { workflowTaskCompletionSizeLimit.set(limits.getWorkflowTaskCompletionSizeLimitError()); + setPayloadErrorLimits(limits.getBlobSizeLimitError(), limits.getMemoSizeLimitError()); } public boolean isPollerAutoscaling() { @@ -95,4 +100,19 @@ public long getWorkflowTaskCompletionSizeLimit() { public void setWorkflowTaskCompletionSizeLimit(long value) { workflowTaskCompletionSizeLimit.set(value); } + + /** + * Returns the namespace's payload and memo error limits, which are {@link + * PayloadErrorLimits#NONE} until the namespace advertises any. + */ + public PayloadErrorLimits getPayloadErrorLimits() { + return payloadErrorLimits.get(); + } + + /** Sets the payload and memo error limits in bytes; a value of 0 or less disables that limit. */ + public void setPayloadErrorLimits(long blob, long memo) { + long blobLimit = Math.max(0, blob); + long memoLimit = Math.max(0, memo); + payloadErrorLimits.set(new PayloadErrorLimits(blobLimit, memoLimit)); + } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/NexusWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/NexusWorker.java index d583ff2281..6f06660121 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/NexusWorker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/NexusWorker.java @@ -9,6 +9,7 @@ import com.uber.m3.tally.Stopwatch; import com.uber.m3.util.Duration; import com.uber.m3.util.ImmutableMap; +import io.grpc.StatusRuntimeException; import io.nexusrpc.OperationState; import io.nexusrpc.handler.HandlerException; import io.temporal.api.failure.v1.Failure; @@ -16,10 +17,14 @@ import io.temporal.api.workflowservice.v1.*; import io.temporal.common.converter.DataConverter; import io.temporal.internal.common.NexusUtil; +import io.temporal.internal.common.PayloadsTooLargeFailure; import io.temporal.internal.common.ProtobufTimeUtils; import io.temporal.internal.concurrent.structured.CancelSource; import io.temporal.internal.logging.LoggerTag; import io.temporal.internal.logging.PrefixedMdc; +import io.temporal.internal.payload.limits.PayloadErrorLimits; +import io.temporal.internal.payload.limits.PayloadLimitViolation; +import io.temporal.internal.payload.limits.PayloadLimitViolationException; import io.temporal.internal.payload.storage.ExternalStorageNotConfiguredException; import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.internal.retryer.GrpcRetryer; @@ -33,6 +38,7 @@ import io.temporal.worker.tuning.PollerBehaviorAutoscaling; import java.util.Collections; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.CancellationException; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; @@ -467,6 +473,22 @@ private boolean handleNexusTask(NexusTask task, Scope metricsScope) { } catch (CancellationException e) { // Absorbed by handle() when this worker is shutting down. throw e; + } catch (StatusRuntimeException e) { + // No Nexus reply field has an error limit today (see PayloadLimitValidatorGenerator), so + // this never runs yet. It is here so that a field classified later fails the task cleanly + // instead of surfacing as a reporting failure. + Optional violation = PayloadLimitViolationException.find(e); + if (!violation.isPresent()) { + logExceptionDuringResultReporting(e, pollResponse, result); + throw e; + } + // The validator already logged the violation. + if (!failed) { + PayloadsTooLargeFailure.count(metricsScope, MetricsType.NEXUS_EXEC_FAILED_COUNTER); + } + sendPayloadsTooLargeFailure( + taskToken, supportsTemporalFailure(pollResponse), metricsScope, violation.get()); + return true; } catch (Exception e) { logExceptionDuringResultReporting(e, pollResponse, result); throw e; @@ -544,6 +566,9 @@ private void sendReply( service .blockingStub() .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) + .withOption( + PayloadErrorLimits.CALL_OPTIONS_KEY, + options.payloadErrorLimits(namespaceCapabilities)) .respondNexusTaskCompleted(request), replyGrpcRetryerOptions); } else { @@ -577,6 +602,9 @@ private void sendReply( service .blockingStub() .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) + .withOption( + PayloadErrorLimits.CALL_OPTIONS_KEY, + options.payloadErrorLimits(namespaceCapabilities)) .respondNexusTaskFailed(failedRequest), replyGrpcRetryerOptions); } else { @@ -620,6 +648,28 @@ private void sendStorageFailure( false); } + /** + * Reports a completion over the payload error limits as a retryable internal handler error. The + * failure carries no payloads, so it bypasses external storage. + */ + private void sendPayloadsTooLargeFailure( + ByteString taskToken, + boolean supportTemporalFailure, + Scope metricsScope, + PayloadLimitViolation violation) { + sendReply( + taskToken, + supportTemporalFailure, + new NexusTaskHandler.Result( + new HandlerException( + HandlerException.ErrorType.INTERNAL, + violation.getMessage(), + PayloadsTooLargeFailure.of(violation), + HandlerException.RetryBehavior.RETRYABLE)), + metricsScope, + false); + } + /** * True when {@code e} is external storage aborting because this worker is shutting down. A * storage driver that genuinely breaks at the same moment is a different thing and must still diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/SingleWorkerOptions.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/SingleWorkerOptions.java index 3d6633dbbe..0f3a0446d5 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/SingleWorkerOptions.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/SingleWorkerOptions.java @@ -9,6 +9,7 @@ import io.temporal.common.converter.GlobalDataConverter; import io.temporal.common.interceptors.WorkerInterceptor; import io.temporal.internal.logging.PrefixedMdc; +import io.temporal.internal.payload.limits.PayloadErrorLimits; import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.worker.PreferredVersionProvider; import io.temporal.worker.WorkerDeploymentOptions; @@ -48,6 +49,7 @@ public static final class Builder { private WorkerDeploymentOptions deploymentOptions; private String workerInstanceKey; private boolean allowActivityHeartbeatDuringShutdown; + private boolean payloadErrorLimitDisabled; private String workerControlTaskQueue; private PreferredVersionProvider preferredVersionProvider; private @Nullable ExternalStorageRunner externalStorageRunner; @@ -79,6 +81,7 @@ private Builder(SingleWorkerOptions options) { this.deploymentOptions = options.getDeploymentOptions(); this.workerInstanceKey = options.getWorkerInstanceKey(); this.allowActivityHeartbeatDuringShutdown = options.getAllowActivityHeartbeatDuringShutdown(); + this.payloadErrorLimitDisabled = options.isPayloadErrorLimitDisabled(); this.workerControlTaskQueue = options.getWorkerControlTaskQueue(); this.preferredVersionProvider = options.getPreferredVersionProvider(); this.externalStorageRunner = options.getExternalStorageRunner(); @@ -186,6 +189,11 @@ public Builder setAllowActivityHeartbeatDuringShutdown( return this; } + public Builder setPayloadErrorLimitDisabled(boolean payloadErrorLimitDisabled) { + this.payloadErrorLimitDisabled = payloadErrorLimitDisabled; + return this; + } + public Builder setWorkerControlTaskQueue(String workerControlTaskQueue) { this.workerControlTaskQueue = workerControlTaskQueue; return this; @@ -259,6 +267,7 @@ public SingleWorkerOptions build() { this.deploymentOptions, this.workerInstanceKey, this.allowActivityHeartbeatDuringShutdown, + this.payloadErrorLimitDisabled, this.workerControlTaskQueue, this.preferredVersionProvider, this.externalStorageRunner, @@ -286,6 +295,7 @@ public SingleWorkerOptions build() { private final WorkerDeploymentOptions deploymentOptions; private final String workerInstanceKey; private final boolean allowActivityHeartbeatDuringShutdown; + private final boolean payloadErrorLimitDisabled; private final String workerControlTaskQueue; private final PreferredVersionProvider preferredVersionProvider; private final @Nullable ExternalStorageRunner externalStorageRunner; @@ -312,6 +322,7 @@ private SingleWorkerOptions( WorkerDeploymentOptions deploymentOptions, String workerInstanceKey, boolean allowActivityHeartbeatDuringShutdown, + boolean payloadErrorLimitDisabled, String workerControlTaskQueue, PreferredVersionProvider preferredVersionProvider, @Nullable ExternalStorageRunner externalStorageRunner, @@ -336,6 +347,7 @@ private SingleWorkerOptions( this.deploymentOptions = deploymentOptions; this.workerInstanceKey = workerInstanceKey; this.allowActivityHeartbeatDuringShutdown = allowActivityHeartbeatDuringShutdown; + this.payloadErrorLimitDisabled = payloadErrorLimitDisabled; this.workerControlTaskQueue = workerControlTaskQueue; this.preferredVersionProvider = preferredVersionProvider; this.externalStorageRunner = externalStorageRunner; @@ -378,6 +390,16 @@ public boolean getAllowActivityHeartbeatDuringShutdown() { return allowActivityHeartbeatDuringShutdown; } + public boolean isPayloadErrorLimitDisabled() { + return payloadErrorLimitDisabled; + } + + public PayloadErrorLimits payloadErrorLimits(NamespaceCapabilities namespaceCapabilities) { + return payloadErrorLimitDisabled + ? PayloadErrorLimits.NONE + : namespaceCapabilities.getPayloadErrorLimits(); + } + public DataConverter getDataConverter() { return dataConverter; } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java index d5cf77f135..6a57e30deb 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java @@ -60,7 +60,8 @@ public SyncActivityWorker( options.getDefaultHeartbeatThrottleInterval(), options.getDataConverter(), heartbeatExecutor, - options.getExternalStorageRunner()); + options.getExternalStorageRunner(), + () -> options.payloadErrorLimits(namespaceCapabilities)); this.taskHandler = new ActivityTaskHandlerImpl( namespace, diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java index 613526fad4..16ca1d27f7 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java @@ -21,9 +21,13 @@ import io.temporal.api.failure.v1.Failure; import io.temporal.api.workflowservice.v1.*; import io.temporal.failure.ApplicationFailure; +import io.temporal.internal.common.PayloadsTooLargeFailure; import io.temporal.internal.common.env.EnvironmentVariableUtils; import io.temporal.internal.logging.LoggerTag; import io.temporal.internal.logging.PrefixedMdc; +import io.temporal.internal.payload.limits.PayloadErrorLimits; +import io.temporal.internal.payload.limits.PayloadLimitViolation; +import io.temporal.internal.payload.limits.PayloadLimitViolationException; import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.internal.payload.storage.StorageOperationMetrics; import io.temporal.internal.payload.visitor.MessageVisitor; @@ -816,6 +820,45 @@ public void handle(WorkflowTask task) throws Exception { workflowTypeScope, workflowStorageTarget(workflowExecution, workflowType), uploadMetrics); + } catch (StatusRuntimeException e) { + Optional violation = + PayloadLimitViolationException.find(e); + if (!violation.isPresent()) { + throw e; + } + releaseReason = SlotReleaseReason.error(e); + // The validator already logged the violation, so the generic reporting failure + // warning is not logged as well. + cache.invalidate( + workflowExecution, + workflowTypeScope, + "Workflow task completion exceeded the payload size limit", + e); + taskFailedCause = + WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_PAYLOADS_TOO_LARGE; + if (currentTask.getAttempt() > 1) { + log.debug( + "Not failing workflow task on non-first attempt. WorkflowId={}, RunId={}, attempt={}", + workflowExecution.getWorkflowId(), + workflowExecution.getRunId(), + currentTask.getAttempt()); + } else { + RespondWorkflowTaskFailedRequest.Builder taskFailedBuilder = + RespondWorkflowTaskFailedRequest.newBuilder() + .setFailure( + payloadsTooLargeFailure( + workflowExecution.getWorkflowId(), violation.get())) + .setCause( + WorkflowTaskFailedCause + .WORKFLOW_TASK_FAILED_CAUSE_PAYLOADS_TOO_LARGE); + sendTaskFailed( + currentTask.getTaskToken(), + taskFailedBuilder, + result.getRequestRetryOptions(), + workflowTypeScope, + workflowStorageTarget(workflowExecution, workflowType), + uploadMetrics); + } } catch (ExternalStorageTaskFailure e) { releaseReason = SlotReleaseReason.error(e); handleReportingFailure( @@ -870,6 +913,9 @@ public void handle(WorkflowTask task) throws Exception { case WORKFLOW_TASK_FAILED_CAUSE_REQUEST_TOO_LARGE: taskFailureType = MetricsTag.TASK_FAILURE_VALUE_REQUEST_TOO_LARGE; break; + case WORKFLOW_TASK_FAILED_CAUSE_PAYLOADS_TOO_LARGE: + taskFailureType = MetricsTag.TASK_FAILURE_VALUE_PAYLOADS_TOO_LARGE; + break; default: taskFailureType = MetricsTag.TASK_FAILURE_VALUE_WORKFLOW_ERROR; } @@ -1077,6 +1123,9 @@ private RespondWorkflowTaskCompletedResponse respondWorkflowTaskCompleted( return service .blockingStub() .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, workflowTypeMetricsScope) + .withOption( + PayloadErrorLimits.CALL_OPTIONS_KEY, + options.payloadErrorLimits(namespaceCapabilities)) .respondWorkflowTaskCompleted(request); } @@ -1226,6 +1275,13 @@ private Failure storageFailure( .exceptionToFailure(applicationFailure); } + private Failure payloadsTooLargeFailure(String workflowId, PayloadLimitViolation violation) { + return options + .getDataConverter() + .withContext(new WorkflowSerializationContext(namespace, workflowId)) + .exceptionToFailure(PayloadsTooLargeFailure.of(violation)); + } + private Failure grpcMessageTooLargeFailure( String workflowId, GrpcMessageTooLargeException e, String messagePrefix) { ApplicationFailure applicationFailure = diff --git a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java index 91f5515ea8..41edc098f6 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java @@ -1128,6 +1128,7 @@ private static SingleWorkerOptions.Builder toSingleWorkerOptions( .setWorkerInstanceKey(workerInstanceKey) .setWorkerControlTaskQueue(workerControlTaskQueue) .setPreferredVersionProvider(options.getPreferredVersionProvider()) + .setPayloadErrorLimitDisabled(options.isPayloadErrorLimitDisabled()) .setLoggerMdc(new PrefixedMdc(factoryOptions.getLoggerTagPrefix())); } diff --git a/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java b/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java index 3346b53142..07b331397e 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java @@ -80,6 +80,7 @@ public static final class Builder { private PollerBehavior activityTaskPollersBehavior; private PollerBehavior nexusTaskPollersBehavior; private boolean allowActivityHeartbeatDuringShutdown; + private boolean disablePayloadErrorLimit; private PreferredVersionProvider preferredVersionProvider; // Track whether the user explicitly configured the pollers for a task type (called either the // max-concurrent-pollers or the poller-behavior setter). A type left unconfigured is eligible @@ -126,6 +127,7 @@ private Builder(WorkerOptions o) { this.activityTaskPollersBehavior = o.activityTaskPollersBehavior; this.nexusTaskPollersBehavior = o.nexusTaskPollersBehavior; this.allowActivityHeartbeatDuringShutdown = o.allowActivityHeartbeatDuringShutdown; + this.disablePayloadErrorLimit = o.disablePayloadErrorLimit; this.preferredVersionProvider = o.preferredVersionProvider; this.workflowTaskPollersConfigured = o.workflowTaskPollersConfigured; this.activityTaskPollersConfigured = o.activityTaskPollersConfigured; @@ -610,6 +612,29 @@ public Builder setAllowActivityHeartbeatDuringShutdown( return this; } + /** + * Disables the worker's enforcement of the namespace's payload and memo error limits. Defaults + * to false. + * + *

By default, a workflow or activity task whose completion contains a payload or memo larger + * than the namespace's error limit is failed by the worker, retryably and with a {@code + * [TMPRL1103]} error log, instead of being sent: a workflow task fails with cause {@code + * PAYLOADS_TOO_LARGE}, and an activity fails with an {@code ApplicationFailure} of type {@code + * PayloadsTooLarge}. When set to true, oversized completions are sent and the server enforces + * the limit. Warnings configured through {@code WorkflowServiceStubsOptions.setPayloadLimits} + * are logged either way. + * + *

Disable enforcement when a proxy between the worker and the server changes payload sizes, + * for example through compression, encryption, or external storage. + * + * @see Blob size limit + * error + */ + public Builder setDisablePayloadErrorLimit(boolean disablePayloadErrorLimit) { + this.disablePayloadErrorLimit = disablePayloadErrorLimit; + return this; + } + /** * Sets a provider that can choose the version recorded by the first non-replay {@link * io.temporal.workflow.Workflow#getVersion(String, int, int)} call for a change ID. @@ -655,6 +680,7 @@ public WorkerOptions build() { activityTaskPollersBehavior, nexusTaskPollersBehavior, allowActivityHeartbeatDuringShutdown, + disablePayloadErrorLimit, preferredVersionProvider, workflowTaskPollersConfigured, activityTaskPollersConfigured, @@ -797,6 +823,7 @@ public WorkerOptions validateAndBuildWithDefaults() { activityTaskPollersBehavior, nexusTaskPollersBehavior, allowActivityHeartbeatDuringShutdown, + disablePayloadErrorLimit, preferredVersionProvider, workflowTaskPollersConfigured, activityTaskPollersConfigured, @@ -834,6 +861,7 @@ public WorkerOptions validateAndBuildWithDefaults() { private final PollerBehavior activityTaskPollersBehavior; private final PollerBehavior nexusTaskPollersBehavior; private final boolean allowActivityHeartbeatDuringShutdown; + private final boolean disablePayloadErrorLimit; private final PreferredVersionProvider preferredVersionProvider; private final boolean workflowTaskPollersConfigured; private final boolean activityTaskPollersConfigured; @@ -870,6 +898,7 @@ private WorkerOptions( PollerBehavior activityTaskPollersBehavior, PollerBehavior nexusTaskPollersBehavior, boolean allowActivityHeartbeatDuringShutdown, + boolean disablePayloadErrorLimit, PreferredVersionProvider preferredVersionProvider, boolean workflowTaskPollersConfigured, boolean activityTaskPollersConfigured, @@ -904,6 +933,7 @@ private WorkerOptions( this.activityTaskPollersBehavior = activityTaskPollersBehavior; this.nexusTaskPollersBehavior = nexusTaskPollersBehavior; this.allowActivityHeartbeatDuringShutdown = allowActivityHeartbeatDuringShutdown; + this.disablePayloadErrorLimit = disablePayloadErrorLimit; this.preferredVersionProvider = preferredVersionProvider; this.workflowTaskPollersConfigured = workflowTaskPollersConfigured; this.activityTaskPollersConfigured = activityTaskPollersConfigured; @@ -1079,6 +1109,14 @@ public boolean getAllowActivityHeartbeatDuringShutdown() { return allowActivityHeartbeatDuringShutdown; } + /** + * @return true if the worker does not enforce the namespace's payload and memo error limits. + * @see Builder#setDisablePayloadErrorLimit(boolean) + */ + public boolean isPayloadErrorLimitDisabled() { + return disablePayloadErrorLimit; + } + @Experimental public PreferredVersionProvider getPreferredVersionProvider() { return preferredVersionProvider; @@ -1120,6 +1158,7 @@ && compare(maxTaskQueueActivitiesPerSecond, that.maxTaskQueueActivitiesPerSecond && Objects.equals(activityTaskPollersBehavior, that.activityTaskPollersBehavior) && Objects.equals(nexusTaskPollersBehavior, that.nexusTaskPollersBehavior) && allowActivityHeartbeatDuringShutdown == that.allowActivityHeartbeatDuringShutdown + && disablePayloadErrorLimit == that.disablePayloadErrorLimit && Objects.equals(preferredVersionProvider, that.preferredVersionProvider); } @@ -1156,6 +1195,7 @@ public int hashCode() { activityTaskPollersBehavior, nexusTaskPollersBehavior, allowActivityHeartbeatDuringShutdown, + disablePayloadErrorLimit, preferredVersionProvider); } @@ -1223,6 +1263,8 @@ public String toString() { + nexusTaskPollersBehavior + ", allowActivityHeartbeatDuringShutdown=" + allowActivityHeartbeatDuringShutdown + + ", disablePayloadErrorLimit=" + + disablePayloadErrorLimit + ", preferredVersionProvider=" + preferredVersionProvider + '}'; diff --git a/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityExecutionContextImplTest.java b/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityExecutionContextImplTest.java index 8cff52cad2..f35ccc8af7 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityExecutionContextImplTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityExecutionContextImplTest.java @@ -13,6 +13,7 @@ import io.temporal.client.WorkflowClient; import io.temporal.common.converter.GlobalDataConverter; import io.temporal.internal.client.external.ManualActivityCompletionClientFactory; +import io.temporal.internal.payload.limits.PayloadErrorLimits; import io.temporal.payload.context.ActivitySerializationContext; import io.temporal.payload.storage.StorageDriverActivityInfo; import io.temporal.payload.storage.StorageDriverTargetInfo; @@ -60,7 +61,8 @@ public void localManualCompletionIncludesActivityTarget() { Duration.ofSeconds(60), Duration.ofSeconds(30), () -> {}, - null); + null, + PayloadErrorLimits.NONE); context.useLocalManualCompletion(); diff --git a/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityTaskHandlerPayloadLimitsTest.java b/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityTaskHandlerPayloadLimitsTest.java new file mode 100644 index 0000000000..943bb8cc6b --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityTaskHandlerPayloadLimitsTest.java @@ -0,0 +1,223 @@ +package io.temporal.internal.activity; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import ch.qos.logback.classic.LoggerContext; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; +import com.google.protobuf.ByteString; +import com.uber.m3.tally.NoopScope; +import io.grpc.Status; +import io.temporal.activity.Activity; +import io.temporal.activity.ActivityInterface; +import io.temporal.api.common.v1.ActivityType; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.common.v1.WorkflowType; +import io.temporal.api.failure.v1.ApplicationFailureInfo; +import io.temporal.api.workflowservice.v1.GetSystemInfoResponse; +import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse; +import io.temporal.api.workflowservice.v1.RespondActivityTaskFailedResponse; +import io.temporal.api.workflowservice.v1.WorkflowServiceGrpc; +import io.temporal.client.ActivityCanceledException; +import io.temporal.client.WorkflowClient; +import io.temporal.common.converter.GlobalDataConverter; +import io.temporal.common.interceptors.WorkerInterceptor; +import io.temporal.internal.payload.limits.LimitClass; +import io.temporal.internal.payload.limits.LimitSeverity; +import io.temporal.internal.payload.limits.PayloadErrorLimits; +import io.temporal.internal.payload.limits.PayloadLimitViolation; +import io.temporal.internal.payload.limits.PayloadLimitViolationException; +import io.temporal.internal.worker.ActivityTask; +import io.temporal.internal.worker.ActivityTaskHandler; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.worker.tuning.SlotPermit; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.slf4j.LoggerFactory; + +/** + * An oversized heartbeat makes the worker fail the activity itself, so the activity's own result + * must not be sent afterwards: the server would answer it with NOT_FOUND. + */ +public class ActivityTaskHandlerPayloadLimitsTest { + + @ActivityInterface + public interface HeartbeatingActivity { + void run(); + } + + public static class HeartbeatingActivityImpl implements HeartbeatingActivity { + private final boolean swallowCancellation; + + HeartbeatingActivityImpl(boolean swallowCancellation) { + this.swallowCancellation = swallowCancellation; + } + + @Override + public void run() { + try { + Activity.getExecutionContext().heartbeat("details"); + } catch (ActivityCanceledException e) { + if (!swallowCancellation) { + throw e; + } + } + } + } + + private ScheduledExecutorService heartbeatExecutor; + private ListAppender executorLogs; + private WorkflowServiceGrpc.WorkflowServiceBlockingStub blockingStub; + private WorkflowClient client; + + @Before + public void setUp() { + heartbeatExecutor = Executors.newScheduledThreadPool(1); + LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory(); + executorLogs = new ListAppender<>(); + executorLogs.setContext(loggerContext); + executorLogs.start(); + loggerContext + .getLogger(ActivityTaskExecutors.ActivityTaskExecutor.class.getName()) + .addAppender(executorLogs); + WorkflowServiceStubs service = mock(WorkflowServiceStubs.class); + blockingStub = mock(WorkflowServiceGrpc.WorkflowServiceBlockingStub.class); + when(service.blockingStub()).thenReturn(blockingStub); + when(blockingStub.withOption(any(), any())).thenReturn(blockingStub); + when(service.getServerCapabilities()) + .thenReturn(() -> GetSystemInfoResponse.Capabilities.getDefaultInstance()); + PayloadLimitViolation violation = + new PayloadLimitViolation("details", LimitClass.BLOB, LimitSeverity.ERROR, 1000, 100); + when(blockingStub.recordActivityTaskHeartbeat(any())) + .thenThrow( + Status.INVALID_ARGUMENT + .withDescription(violation.getMessage()) + .withCause(new PayloadLimitViolationException(violation)) + .asRuntimeException()); + client = mock(WorkflowClient.class); + when(client.getWorkflowServiceStubs()).thenReturn(service); + } + + @After + public void tearDown() { + heartbeatExecutor.shutdownNow(); + ((LoggerContext) LoggerFactory.getILoggerFactory()) + .getLogger(ActivityTaskExecutors.ActivityTaskExecutor.class.getName()) + .detachAppender(executorLogs); + } + + @Test + public void aRethrownCancellationSendsNothing() { + when(blockingStub.respondActivityTaskFailed(any())) + .thenReturn(RespondActivityTaskFailedResponse.getDefaultInstance()); + + ActivityTaskHandler.Result result = runActivity(new HeartbeatingActivityImpl(false)); + + verify(blockingStub).respondActivityTaskFailed(any()); + assertSendsNothing(result); + assertNotLoggedAsCanceled(); + } + + @Test + public void aSwallowedCancellationSendsNothing() { + when(blockingStub.respondActivityTaskFailed(any())) + .thenReturn(RespondActivityTaskFailedResponse.getDefaultInstance()); + + ActivityTaskHandler.Result result = runActivity(new HeartbeatingActivityImpl(true)); + + assertSendsNothing(result); + } + + @Test + public void aRethrownCancellationReportsTheFailureIfTheHeartbeatCouldNotReportIt() { + // FAILED_PRECONDITION is not retried, so the heartbeat's report fails at once. + when(blockingStub.respondActivityTaskFailed(any())) + .thenThrow(Status.FAILED_PRECONDITION.asRuntimeException()); + + ActivityTaskHandler.Result result = runActivity(new HeartbeatingActivityImpl(false)); + + assertReportsPayloadsTooLarge(result); + assertNotLoggedAsCanceled(); + } + + @Test + public void aSwallowedCancellationReportsTheFailureIfTheHeartbeatCouldNotReportIt() { + when(blockingStub.respondActivityTaskFailed(any())) + .thenThrow(Status.FAILED_PRECONDITION.asRuntimeException()); + + ActivityTaskHandler.Result result = runActivity(new HeartbeatingActivityImpl(true)); + + assertReportsPayloadsTooLarge(result); + } + + /** Asserts that the activity reports the PayloadsTooLarge failure rather than its own outcome. */ + private static void assertReportsPayloadsTooLarge(ActivityTaskHandler.Result result) { + assertNull(result.getTaskCompleted()); + assertNull(result.getTaskCanceled()); + assertNotNull(result.getTaskFailed()); + ApplicationFailureInfo info = + result.getTaskFailed().getTaskFailedRequest().getFailure().getApplicationFailureInfo(); + assertEquals("PayloadsTooLarge", info.getType()); + assertFalse(info.getNonRetryable()); + } + + /** Asserts that the activity, which was failed, is not logged as canceled. */ + private void assertNotLoggedAsCanceled() { + for (ILoggingEvent event : executorLogs.list) { + assertFalse(event.getFormattedMessage(), event.getFormattedMessage().contains("canceled")); + } + } + + private ActivityTaskHandler.Result runActivity(HeartbeatingActivity activity) { + ActivityExecutionContextFactoryImpl factory = + new ActivityExecutionContextFactoryImpl( + client, + "test-identity", + "test-namespace", + Duration.ofSeconds(60), + Duration.ofSeconds(30), + GlobalDataConverter.get(), + heartbeatExecutor, + null, + () -> new PayloadErrorLimits(100, 100)); + ActivityTaskHandlerImpl handler = + new ActivityTaskHandlerImpl( + "test-namespace", + "test-task-queue", + GlobalDataConverter.get(), + factory, + new WorkerInterceptor[0], + null); + handler.registerActivityImplementations(new Object[] {activity}); + PollActivityTaskQueueResponse task = + PollActivityTaskQueueResponse.newBuilder() + .setTaskToken(ByteString.copyFrom("token", StandardCharsets.UTF_8)) + .setActivityId("activity-id") + .setActivityType(ActivityType.newBuilder().setName("Run")) + .setWorkflowType(WorkflowType.newBuilder().setName("Workflow")) + .setWorkflowExecution( + WorkflowExecution.newBuilder().setWorkflowId("workflow-id").setRunId("run-id")) + .build(); + return handler.handle( + new ActivityTask(task, new SlotPermit(), () -> {}), new NoopScope(), false); + } + + private static void assertSendsNothing(ActivityTaskHandler.Result result) { + assertNull(result.getTaskCompleted()); + assertNull(result.getTaskFailed()); + assertNull(result.getTaskCanceled()); + assertFalse(result.isManualCompletion()); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/activity/HeartbeatContextImplTest.java b/temporal-sdk/src/test/java/io/temporal/internal/activity/HeartbeatContextImplTest.java index 19fac4d71a..e57e647ee2 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/activity/HeartbeatContextImplTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/activity/HeartbeatContextImplTest.java @@ -1,8 +1,11 @@ package io.temporal.internal.activity; import static org.junit.Assert.*; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.*; +import static org.mockito.Mockito.times; import com.uber.m3.tally.NoopScope; import io.grpc.Status; @@ -10,8 +13,11 @@ import io.temporal.activity.ActivityInfo; import io.temporal.api.common.v1.Payload; import io.temporal.api.enums.v1.TimeoutType; +import io.temporal.api.workflowservice.v1.GetSystemInfoResponse; import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatRequest; import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatResponse; +import io.temporal.api.workflowservice.v1.RespondActivityTaskFailedRequest; +import io.temporal.api.workflowservice.v1.RespondActivityTaskFailedResponse; import io.temporal.api.workflowservice.v1.WorkflowServiceGrpc; import io.temporal.client.ActivityCanceledException; import io.temporal.client.ActivityCompletionException; @@ -19,6 +25,11 @@ import io.temporal.common.CancellationToken; import io.temporal.common.converter.GlobalDataConverter; import io.temporal.failure.TimeoutFailure; +import io.temporal.internal.payload.limits.LimitClass; +import io.temporal.internal.payload.limits.LimitSeverity; +import io.temporal.internal.payload.limits.PayloadErrorLimits; +import io.temporal.internal.payload.limits.PayloadLimitViolation; +import io.temporal.internal.payload.limits.PayloadLimitViolationException; import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.payload.storage.ExternalStorage; import io.temporal.payload.storage.StorageDriver; @@ -63,6 +74,8 @@ public void setUp() { blockingStub = mock(WorkflowServiceGrpc.WorkflowServiceBlockingStub.class); when(service.blockingStub()).thenReturn(blockingStub); when(blockingStub.withOption(any(), any())).thenReturn(blockingStub); + when(service.getServerCapabilities()) + .thenReturn(() -> GetSystemInfoResponse.Capabilities.getDefaultInstance()); } @After @@ -346,7 +359,8 @@ public void factoryCancelByTaskTokenCompletesCancellationToken() { Duration.ofSeconds(30), GlobalDataConverter.get(), heartbeatExecutor, - null); + null, + () -> PayloadErrorLimits.NONE); ActivityInfoInternal info = activityInfoWithHeartbeatTimeout(Duration.ofSeconds(10)); InternalActivityExecutionContext context = @@ -362,6 +376,198 @@ public void factoryCancelByTaskTokenCompletesCancellationToken() { assertFalse(factory.cleanupContext(new byte[] {1, 2, 3}, true)); } + @Test + public void aHeartbeatOverThePayloadErrorLimitFailsTheActivityAndCancelsIt() { + PayloadLimitViolation violation = + new PayloadLimitViolation("details", LimitClass.BLOB, LimitSeverity.ERROR, 1000, 100); + when(blockingStub.recordActivityTaskHeartbeat(any())) + .thenThrow( + Status.INVALID_ARGUMENT + .withDescription(violation.getMessage()) + .withCause(new PayloadLimitViolationException(violation)) + .asRuntimeException()); + when(blockingStub.respondActivityTaskFailed(any())) + .thenReturn(RespondActivityTaskFailedResponse.getDefaultInstance()); + HeartbeatContextImpl ctx = + createHeartbeatContext( + activityInfoWithHeartbeatTimeout(Duration.ZERO), new PayloadErrorLimits(100, 100)); + + assertThrows(ActivityCanceledException.class, () -> ctx.heartbeat("details")); + + ArgumentCaptor sent = + ArgumentCaptor.forClass(RespondActivityTaskFailedRequest.class); + verify(blockingStub).respondActivityTaskFailed(sent.capture()); + assertEquals(violation.getMessage(), sent.getValue().getFailure().getMessage()); + assertEquals( + "PayloadsTooLarge", sent.getValue().getFailure().getApplicationFailureInfo().getType()); + assertFalse(sent.getValue().getFailure().getApplicationFailureInfo().getNonRetryable()); + assertTrue(ctx.getCancellationToken().isCancellationRequested()); + assertTrue(ctx.awaitTaskReported()); + } + + @Test + public void aFailedReportAfterAnOversizedHeartbeatLeavesTheViolationForTheActivity() { + PayloadLimitViolation violation = + new PayloadLimitViolation("details", LimitClass.BLOB, LimitSeverity.ERROR, 1000, 100); + when(blockingStub.recordActivityTaskHeartbeat(any())) + .thenThrow( + Status.INVALID_ARGUMENT + .withDescription(violation.getMessage()) + .withCause(new PayloadLimitViolationException(violation)) + .asRuntimeException()); + // FAILED_PRECONDITION is not retried, so the report fails at once. + when(blockingStub.respondActivityTaskFailed(any())) + .thenThrow(Status.FAILED_PRECONDITION.asRuntimeException()); + HeartbeatContextImpl ctx = + createHeartbeatContext( + activityInfoWithHeartbeatTimeout(Duration.ZERO), new PayloadErrorLimits(100, 100)); + + assertThrows(ActivityCanceledException.class, () -> ctx.heartbeat("details")); + + assertFalse(ctx.awaitTaskReported()); + assertSame(violation, ctx.getPayloadLimitViolation()); + } + + @Test + public void aTransientFailureToReportAnOversizedHeartbeatIsRetried() { + PayloadLimitViolation violation = + new PayloadLimitViolation("details", LimitClass.BLOB, LimitSeverity.ERROR, 1000, 100); + when(blockingStub.recordActivityTaskHeartbeat(any())) + .thenThrow( + Status.INVALID_ARGUMENT + .withDescription(violation.getMessage()) + .withCause(new PayloadLimitViolationException(violation)) + .asRuntimeException()); + when(blockingStub.respondActivityTaskFailed(any())) + .thenThrow(Status.UNAVAILABLE.asRuntimeException()) + .thenReturn(RespondActivityTaskFailedResponse.getDefaultInstance()); + HeartbeatContextImpl ctx = + createHeartbeatContext( + activityInfoWithHeartbeatTimeout(Duration.ZERO), new PayloadErrorLimits(100, 100)); + + assertThrows(ActivityCanceledException.class, () -> ctx.heartbeat("details")); + + verify(blockingStub, times(2)).respondActivityTaskFailed(any()); + assertTrue(ctx.awaitTaskReported()); + assertSame(violation, ctx.getPayloadLimitViolation()); + } + + @Test + public void heartbeatsAfterAnOversizedOneAreNotSent() { + PayloadLimitViolation violation = + new PayloadLimitViolation("details", LimitClass.BLOB, LimitSeverity.ERROR, 1000, 100); + when(blockingStub.recordActivityTaskHeartbeat(any())) + .thenThrow( + Status.INVALID_ARGUMENT + .withDescription(violation.getMessage()) + .withCause(new PayloadLimitViolationException(violation)) + .asRuntimeException()); + when(blockingStub.respondActivityTaskFailed(any())) + .thenReturn(RespondActivityTaskFailedResponse.getDefaultInstance()); + // A short throttle interval, so that a throttled heartbeat would be sent during the test. + HeartbeatContextImpl ctx = + new HeartbeatContextImpl( + service, + "test-namespace", + activityInfoWithHeartbeatTimeout(Duration.ZERO), + GlobalDataConverter.get(), + heartbeatExecutor, + new NoopScope(), + "test-identity", + Duration.ofMillis(50), + Duration.ofMillis(50), + null, + new PayloadErrorLimits(100, 100), + TEST_BUFFER_MILLIS); + + assertThrows(ActivityCanceledException.class, () -> ctx.heartbeat("details")); + // An activity that swallows the cancellation and keeps heartbeating must not fail the task + // a second time. + assertThrows(ActivityCanceledException.class, () -> ctx.heartbeat("details")); + + verify(blockingStub, after(500).times(1)).recordActivityTaskHeartbeat(any()); + verify(blockingStub, times(1)).respondActivityTaskFailed(any()); + } + + @Test + public void anOversizedHeartbeatIsReportedWithoutHoldingTheHeartbeatLock() throws Exception { + PayloadLimitViolation violation = + new PayloadLimitViolation("details", LimitClass.BLOB, LimitSeverity.ERROR, 1000, 100); + when(blockingStub.recordActivityTaskHeartbeat(any())) + .thenThrow( + Status.INVALID_ARGUMENT + .withDescription(violation.getMessage()) + .withCause(new PayloadLimitViolationException(violation)) + .asRuntimeException()); + CountDownLatch reportStarted = new CountDownLatch(1); + CountDownLatch releaseReport = new CountDownLatch(1); + when(blockingStub.respondActivityTaskFailed(any())) + .thenAnswer( + invocation -> { + reportStarted.countDown(); + releaseReport.await(); + return RespondActivityTaskFailedResponse.getDefaultInstance(); + }); + HeartbeatContextImpl ctx = + createHeartbeatContext( + activityInfoWithHeartbeatTimeout(Duration.ZERO), new PayloadErrorLimits(100, 100)); + + CompletableFuture heartbeat = + CompletableFuture.supplyAsync( + () -> { + try { + ctx.heartbeat("details"); + return null; + } catch (Throwable e) { + return e; + } + }); + try { + assertTrue(reportStarted.await(10, TimeUnit.SECONDS)); + // While the failure is still being sent, the cancellation is already visible and the lock + // is free for other callers. + assertTrue(ctx.getCancellationToken().isCancellationRequested()); + CompletableFuture lockedCall = + CompletableFuture.supplyAsync(ctx::getLatestHeartbeatDetails); + assertEquals("details", lockedCall.get(10, TimeUnit.SECONDS)); + } finally { + releaseReport.countDown(); + } + + assertTrue(heartbeat.get(10, TimeUnit.SECONDS) instanceof ActivityCanceledException); + assertTrue(ctx.awaitTaskReported()); + } + + @Test + public void heartbeatsCarryThePayloadErrorLimits() { + when(blockingStub.recordActivityTaskHeartbeat(any())) + .thenReturn(RecordActivityTaskHeartbeatResponse.getDefaultInstance()); + PayloadErrorLimits limits = new PayloadErrorLimits(100, 100); + HeartbeatContextImpl ctx = + createHeartbeatContext(activityInfoWithHeartbeatTimeout(Duration.ZERO), limits); + + ctx.heartbeat("details"); + + verify(blockingStub).withOption(PayloadErrorLimits.CALL_OPTIONS_KEY, limits); + } + + private HeartbeatContextImpl createHeartbeatContext( + ActivityInfo info, PayloadErrorLimits payloadErrorLimits) { + return new HeartbeatContextImpl( + service, + "test-namespace", + info, + GlobalDataConverter.get(), + heartbeatExecutor, + new NoopScope(), + "test-identity", + Duration.ofSeconds(60), + Duration.ofSeconds(30), + null, + payloadErrorLimits, + TEST_BUFFER_MILLIS); + } + private HeartbeatContextImpl createHeartbeatContext(ActivityInfo info) { return createHeartbeatContext(info, Duration.ofSeconds(60), Duration.ofSeconds(30)); } @@ -381,6 +587,7 @@ private HeartbeatContextImpl createHeartbeatContext( maxHeartbeatThrottleInterval, defaultHeartbeatThrottleInterval, null, + PayloadErrorLimits.NONE, TEST_BUFFER_MILLIS); } @@ -515,6 +722,7 @@ private HeartbeatContextImpl createHeartbeatContextWithStorage(StorageDriver dri Duration.ofSeconds(60), OFFLOAD_INTERVAL, runner, + PayloadErrorLimits.NONE, TEST_BUFFER_MILLIS); } diff --git a/temporal-sdk/src/test/java/io/temporal/internal/worker/ActivityWorkerPayloadLimitsTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/ActivityWorkerPayloadLimitsTest.java new file mode 100644 index 0000000000..6264e8ffd2 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/worker/ActivityWorkerPayloadLimitsTest.java @@ -0,0 +1,219 @@ +package io.temporal.internal.worker; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.google.protobuf.ByteString; +import com.uber.m3.tally.RootScopeBuilder; +import com.uber.m3.tally.Scope; +import com.uber.m3.util.ImmutableMap; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import io.temporal.api.common.v1.ActivityType; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.common.v1.WorkflowType; +import io.temporal.api.failure.v1.Failure; +import io.temporal.api.namespace.v1.NamespaceInfo; +import io.temporal.api.workflowservice.v1.GetSystemInfoResponse; +import io.temporal.api.workflowservice.v1.PollActivityTaskQueueRequest; +import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse; +import io.temporal.api.workflowservice.v1.RespondActivityTaskCompletedRequest; +import io.temporal.api.workflowservice.v1.RespondActivityTaskCompletedResponse; +import io.temporal.api.workflowservice.v1.RespondActivityTaskFailedRequest; +import io.temporal.api.workflowservice.v1.RespondActivityTaskFailedResponse; +import io.temporal.api.workflowservice.v1.WorkflowServiceGrpc; +import io.temporal.common.reporter.TestStatsReporter; +import io.temporal.internal.payload.limits.LimitClass; +import io.temporal.internal.payload.limits.LimitSeverity; +import io.temporal.internal.payload.limits.PayloadErrorLimits; +import io.temporal.internal.payload.limits.PayloadLimitViolation; +import io.temporal.internal.payload.limits.PayloadLimitViolationException; +import io.temporal.serviceclient.MetricsTag; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.worker.MetricsType; +import io.temporal.worker.tuning.FixedSizeSlotSupplier; +import io.temporal.worker.tuning.PollerBehaviorSimpleMaximum; +import java.nio.charset.StandardCharsets; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; +import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.stubbing.Answer; + +public class ActivityWorkerPayloadLimitsTest { + + private static final String ERROR_MESSAGE = + "[TMPRL1103] Attempted to upload payloads with size that exceeded the error limit."; + + private final TestStatsReporter reporter = new TestStatsReporter(); + + @Test + public void aResultOverThePayloadErrorLimitFailsTheActivityRetryably() throws Exception { + ArgumentCaptor sent = + ArgumentCaptor.forClass(RespondActivityTaskFailedRequest.class); + CountDownLatch failed = new CountDownLatch(1); + + runOneActivityTask( + payloadErrorLimits(), + blockingStub -> { + when(blockingStub.respondActivityTaskCompleted( + any(RespondActivityTaskCompletedRequest.class))) + .thenThrow(payloadLimitViolation()); + when(blockingStub.respondActivityTaskFailed(any(RespondActivityTaskFailedRequest.class))) + .thenAnswer( + (Answer) + invocation -> { + failed.countDown(); + return RespondActivityTaskFailedResponse.getDefaultInstance(); + }); + }, + failed, + blockingStub -> verify(blockingStub).respondActivityTaskFailed(sent.capture())); + + Failure failure = sent.getValue().getFailure(); + assertEquals(ERROR_MESSAGE, failure.getMessage()); + assertEquals("PayloadsTooLarge", failure.getApplicationFailureInfo().getType()); + assertFalse(failure.getApplicationFailureInfo().getNonRetryable()); + reporter.assertCounter( + MetricsType.ACTIVITY_EXEC_FAILED_COUNTER, + new ImmutableMap.Builder() + .put("worker_type", "ActivityWorker") + .put("workflow_type", "Workflow") + .put("activity_type", "Activity") + .put(MetricsTag.TASK_FAILURE_TYPE, MetricsTag.TASK_FAILURE_VALUE_PAYLOADS_TOO_LARGE) + .build(), + 1); + } + + @Test + public void completionsCarryTheNamespacePayloadErrorLimits() throws Exception { + NamespaceCapabilities capabilities = payloadErrorLimits(); + CountDownLatch completed = new CountDownLatch(1); + + runOneActivityTask( + capabilities, + blockingStub -> + when(blockingStub.respondActivityTaskCompleted( + any(RespondActivityTaskCompletedRequest.class))) + .thenAnswer( + (Answer) + invocation -> { + completed.countDown(); + return RespondActivityTaskCompletedResponse.getDefaultInstance(); + }), + completed, + blockingStub -> + verify(blockingStub) + .withOption( + PayloadErrorLimits.CALL_OPTIONS_KEY, capabilities.getPayloadErrorLimits())); + } + + private static NamespaceCapabilities payloadErrorLimits() { + NamespaceCapabilities capabilities = new NamespaceCapabilities(); + capabilities.setFromLimits( + NamespaceInfo.Limits.newBuilder() + .setBlobSizeLimitError(100) + .setMemoSizeLimitError(100) + .build()); + return capabilities; + } + + /** Returns what the payload limits interceptor throws for a completion over an error limit. */ + private static StatusRuntimeException payloadLimitViolation() { + PayloadLimitViolation violation = + new PayloadLimitViolation("result", LimitClass.BLOB, LimitSeverity.ERROR, 1000, 100); + return Status.INVALID_ARGUMENT + .withDescription(violation.getMessage()) + .withCause(new PayloadLimitViolationException(violation)) + .asRuntimeException(); + } + + /** + * Polls one activity task whose handler completes it, runs {@code stubSetup}, waits for {@code + * done}, then verifies. + */ + @SuppressWarnings("deprecation") + private void runOneActivityTask( + NamespaceCapabilities namespaceCapabilities, + Consumer stubSetup, + CountDownLatch done, + Consumer verification) + throws Exception { + WorkflowServiceStubs service = mock(WorkflowServiceStubs.class); + when(service.getServerCapabilities()) + .thenReturn(() -> GetSystemInfoResponse.Capabilities.getDefaultInstance()); + WorkflowServiceGrpc.WorkflowServiceBlockingStub blockingStub = + mock(WorkflowServiceGrpc.WorkflowServiceBlockingStub.class); + when(service.blockingStub()).thenReturn(blockingStub); + when(blockingStub.withOption(any(), any())).thenReturn(blockingStub); + + PollActivityTaskQueueResponse task = + PollActivityTaskQueueResponse.newBuilder() + .setTaskToken(ByteString.copyFrom("token", StandardCharsets.UTF_8)) + .setActivityId("activity-id") + .setActivityType(ActivityType.newBuilder().setName("Activity")) + .setWorkflowType(WorkflowType.newBuilder().setName("Workflow")) + .setWorkflowExecution( + WorkflowExecution.newBuilder().setWorkflowId("workflow-id").setRunId("run-id")) + .build(); + CountDownLatch blockPolls = new CountDownLatch(1); + when(blockingStub.pollActivityTaskQueue(any(PollActivityTaskQueueRequest.class))) + .thenReturn(task) + .thenAnswer( + (Answer) + invocation -> { + blockPolls.await(); + return null; + }); + stubSetup.accept(blockingStub); + + ActivityTaskHandler handler = mock(ActivityTaskHandler.class); + when(handler.isAnyTypeSupported()).thenReturn(true); + when(handler.handle(any(ActivityTask.class), any(Scope.class), anyBoolean())) + .thenReturn( + new ActivityTaskHandler.Result( + "activity-id", + RespondActivityTaskCompletedRequest.getDefaultInstance(), + null, + null, + false)); + + Scope metricsScope = + new RootScopeBuilder() + .reporter(reporter) + .reportEvery(com.uber.m3.util.Duration.ofMillis(1)); + ActivityWorker worker = + new ActivityWorker( + service, + "default", + "task_queue", + 1.0, + SingleWorkerOptions.newBuilder() + .setIdentity("test_identity") + .setBuildId(UUID.randomUUID().toString()) + .setWorkerInstanceKey(UUID.randomUUID().toString()) + .setPollerOptions( + PollerOptions.newBuilder() + .setPollerBehavior(new PollerBehaviorSimpleMaximum(1)) + .build()) + .setMetricsScope(metricsScope) + .build(), + handler, + new FixedSizeSlotSupplier<>(10), + namespaceCapabilities); + + assertTrue(worker.start()); + assertTrue(done.await(10, TimeUnit.SECONDS)); + worker.shutdown(new ShutdownManager(), false).get(); + blockPolls.countDown(); + verification.accept(blockingStub); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/worker/NexusWorkerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/NexusWorkerTest.java index f299d4c5d6..99f99a08f6 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/worker/NexusWorkerTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/worker/NexusWorkerTest.java @@ -3,6 +3,7 @@ import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; @@ -16,10 +17,14 @@ import com.uber.m3.tally.NoopScope; import com.uber.m3.tally.RootScopeBuilder; import com.uber.m3.tally.Scope; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; import io.temporal.api.common.v1.Payload; import io.temporal.api.common.v1.Payloads; +import io.temporal.api.enums.v1.NexusHandlerErrorRetryBehavior; import io.temporal.api.failure.v1.ApplicationFailureInfo; import io.temporal.api.failure.v1.Failure; +import io.temporal.api.namespace.v1.NamespaceInfo; import io.temporal.api.nexus.v1.Request; import io.temporal.api.nexus.v1.Response; import io.temporal.api.nexus.v1.StartOperationRequest; @@ -28,12 +33,18 @@ import io.temporal.api.workflowservice.v1.GetSystemInfoResponse; import io.temporal.api.workflowservice.v1.PollNexusTaskQueueRequest; import io.temporal.api.workflowservice.v1.PollNexusTaskQueueResponse; +import io.temporal.api.workflowservice.v1.RespondNexusTaskCompletedRequest; import io.temporal.api.workflowservice.v1.RespondNexusTaskFailedRequest; import io.temporal.api.workflowservice.v1.ShutdownWorkerRequest; import io.temporal.api.workflowservice.v1.ShutdownWorkerResponse; import io.temporal.api.workflowservice.v1.WorkflowServiceGrpc; import io.temporal.common.converter.DefaultDataConverter; import io.temporal.common.reporter.TestStatsReporter; +import io.temporal.internal.payload.limits.LimitClass; +import io.temporal.internal.payload.limits.LimitSeverity; +import io.temporal.internal.payload.limits.PayloadErrorLimits; +import io.temporal.internal.payload.limits.PayloadLimitViolation; +import io.temporal.internal.payload.limits.PayloadLimitViolationException; import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.payload.storage.ExternalStorage; import io.temporal.payload.storage.StorageDriver; @@ -60,6 +71,7 @@ import java.util.concurrent.atomic.AtomicInteger; import javax.annotation.Nonnull; import org.junit.Test; +import org.mockito.ArgumentCaptor; import org.mockito.stubbing.Answer; public class NexusWorkerTest { @@ -223,29 +235,132 @@ public void aHandlerFailureFollowedByAStorageFailureCountsOneExecFailure() throw execFailedTags(MetricsTag.TASK_FAILURE_VALUE_HANDLER_ERROR_INTERNAL)); } + /** + * A completion over the namespace's payload error limit is reported as a retryable handler error + * instead, and counted once as PayloadsTooLarge. + */ + @Test + public void aCompletionOverThePayloadErrorLimitFailsTheTaskRetryably() throws Exception { + Fixture fixture = new Fixture(metricsScope, payloadErrorLimits(), syncSuccess()); + when(fixture.blockingStub.respondNexusTaskCompleted( + any(RespondNexusTaskCompletedRequest.class))) + .thenThrow(payloadLimitViolation()); + CountDownLatch reported = reportedLatch(fixture); + + try { + assertTrue(fixture.worker.start()); + assertTrue("the violation must be reported", reported.await(10, TimeUnit.SECONDS)); + } finally { + shutdown(fixture); + } + + ArgumentCaptor sent = + ArgumentCaptor.forClass(RespondNexusTaskFailedRequest.class); + verify(fixture.blockingStub).respondNexusTaskFailed(sent.capture()); + Failure failure = sent.getValue().getFailure(); + assertEquals("INTERNAL", failure.getNexusHandlerFailureInfo().getType()); + assertEquals( + NexusHandlerErrorRetryBehavior.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_RETRYABLE, + failure.getNexusHandlerFailureInfo().getRetryBehavior()); + assertEquals("PayloadsTooLarge", failure.getCause().getApplicationFailureInfo().getType()); + Eventually.assertEventually( + Duration.ofSeconds(10), + () -> + reporter.assertCounter( + MetricsType.NEXUS_EXEC_FAILED_COUNTER, + execFailedTags(MetricsTag.TASK_FAILURE_VALUE_PAYLOADS_TOO_LARGE), + 1)); + } + + /** + * An operation failure is already counted under its own reason. Its report being over the payload + * error limit must not count the same task a second time. + */ + @Test + public void anOperationFailureOverThePayloadErrorLimitCountsOneExecFailure() throws Exception { + Fixture fixture = new Fixture(metricsScope, payloadErrorLimits(), operationFailure()); + when(fixture.blockingStub.respondNexusTaskCompleted( + any(RespondNexusTaskCompletedRequest.class))) + .thenThrow(payloadLimitViolation()); + CountDownLatch reported = reportedLatch(fixture); + + try { + assertTrue(fixture.worker.start()); + assertTrue("the violation must be reported", reported.await(10, TimeUnit.SECONDS)); + } finally { + shutdown(fixture); + } + + Eventually.assertEventually( + Duration.ofSeconds(10), + () -> + reporter.assertCounter( + MetricsType.NEXUS_EXEC_FAILED_COUNTER, + execFailedTags(MetricsTag.TASK_FAILURE_VALUE_OPERATION_FAILED), + 1)); + reporter.assertNoMetric( + MetricsType.NEXUS_EXEC_FAILED_COUNTER, + execFailedTags(MetricsTag.TASK_FAILURE_VALUE_PAYLOADS_TOO_LARGE)); + } + + @Test + public void repliesCarryTheNamespacePayloadErrorLimits() throws Exception { + NamespaceCapabilities capabilities = payloadErrorLimits(); + Fixture fixture = new Fixture(new NoopScope(), capabilities, syncSuccess()); + CountDownLatch completed = new CountDownLatch(1); + when(fixture.blockingStub.respondNexusTaskCompleted( + any(RespondNexusTaskCompletedRequest.class))) + .thenAnswer( + (Answer) + invocation -> { + completed.countDown(); + return null; + }); + + try { + assertTrue(fixture.worker.start()); + assertTrue("the task must complete", completed.await(10, TimeUnit.SECONDS)); + } finally { + shutdown(fixture); + } + + PayloadErrorLimits limits = capabilities.getPayloadErrorLimits(); + assertNotNull(limits); + verify(fixture.blockingStub).withOption(PayloadErrorLimits.CALL_OPTIONS_KEY, limits); + } + /** A worker wired to {@code driver}, polling exactly one nexus task that returns a payload. */ private static final class Fixture { final NexusWorker worker; final WorkflowServiceGrpc.WorkflowServiceBlockingStub blockingStub; Fixture(StorageDriver driver) throws Exception { - this(driver, null, new NoopScope(), syncSuccess()); + this(driver, null, new NoopScope(), syncSuccess(), new NamespaceCapabilities()); } Fixture(StorageDriver driver, Payload inboundPayload) throws Exception { - this(driver, inboundPayload, new NoopScope(), syncSuccess()); + this(driver, inboundPayload, new NoopScope(), syncSuccess(), new NamespaceCapabilities()); } Fixture(StorageDriver driver, Scope metricsScope, NexusTaskHandler.Result handlerResult) throws Exception { - this(driver, null, metricsScope, handlerResult); + this(driver, null, metricsScope, handlerResult, new NamespaceCapabilities()); + } + + Fixture( + Scope metricsScope, + NamespaceCapabilities namespaceCapabilities, + NexusTaskHandler.Result handlerResult) + throws Exception { + this(null, null, metricsScope, handlerResult, namespaceCapabilities); } private Fixture( StorageDriver driver, Payload inboundPayload, Scope metricsScope, - NexusTaskHandler.Result handlerResult) + NexusTaskHandler.Result handlerResult, + NamespaceCapabilities namespaceCapabilities) throws Exception { WorkflowServiceStubs service = mock(WorkflowServiceStubs.class); when(service.getServerCapabilities()) @@ -309,7 +424,7 @@ private Fixture( handler, DefaultDataConverter.newDefaultInstance(), new FixedSizeSlotSupplier<>(10), - new NamespaceCapabilities()); + namespaceCapabilities); } } @@ -352,6 +467,31 @@ private static Map execFailedTags(String failureReason) { failureReason); } + private static NamespaceCapabilities payloadErrorLimits() { + NamespaceCapabilities capabilities = new NamespaceCapabilities(); + capabilities.setFromLimits( + NamespaceInfo.Limits.newBuilder() + .setBlobSizeLimitError(100) + .setMemoSizeLimitError(100) + .build()); + return capabilities; + } + + /** Returns what the payload limits interceptor throws for a reply over an error limit. */ + private static StatusRuntimeException payloadLimitViolation() { + PayloadLimitViolation violation = + new PayloadLimitViolation( + "response.start_operation.sync_success.payload", + LimitClass.BLOB, + LimitSeverity.ERROR, + 1000, + 100); + return Status.INVALID_ARGUMENT + .withDescription(violation.getMessage()) + .withCause(new PayloadLimitViolationException(violation)) + .asRuntimeException(); + } + private static NexusTaskHandler.Result syncSuccess() { Payload result = Payload.newBuilder().setData(ByteString.copyFrom("a result", UTF_8)).build(); return new NexusTaskHandler.Result( diff --git a/temporal-sdk/src/test/java/io/temporal/internal/worker/PayloadErrorLimitsWorkerOptionsTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/PayloadErrorLimitsWorkerOptionsTest.java new file mode 100644 index 0000000000..6526c7a940 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/worker/PayloadErrorLimitsWorkerOptionsTest.java @@ -0,0 +1,56 @@ +package io.temporal.internal.worker; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; + +import io.temporal.api.namespace.v1.NamespaceInfo; +import io.temporal.internal.payload.limits.PayloadErrorLimits; +import org.junit.Test; + +public class PayloadErrorLimitsWorkerOptionsTest { + + @Test + public void theWorkerAttachesTheNamespaceErrorLimits() { + NamespaceCapabilities capabilities = limits(2_000, 3_000); + PayloadErrorLimits limits = + SingleWorkerOptions.newBuilder().build().payloadErrorLimits(capabilities); + assertEquals(2_000, limits.getBlob()); + assertEquals(3_000, limits.getMemo()); + assertSame(capabilities.getPayloadErrorLimits(), limits); + } + + @Test + public void disablingEnforcementAttachesNoErrorLimits() { + assertSame( + PayloadErrorLimits.NONE, + SingleWorkerOptions.newBuilder() + .setPayloadErrorLimitDisabled(true) + .build() + .payloadErrorLimits(limits(2_000, 3_000))); + } + + @Test + public void aNamespaceWithoutErrorLimitsAttachesNone() { + assertSame(PayloadErrorLimits.NONE, new NamespaceCapabilities().getPayloadErrorLimits()); + PayloadErrorLimits limits = limits(0, 0).getPayloadErrorLimits(); + assertEquals(0, limits.getBlob()); + assertEquals(0, limits.getMemo()); + } + + @Test + public void negativeErrorLimitsDisableThatLimit() { + PayloadErrorLimits limits = limits(-1, 3_000).getPayloadErrorLimits(); + assertEquals(0, limits.getBlob()); + assertEquals(3_000, limits.getMemo()); + } + + private static NamespaceCapabilities limits(long blob, long memo) { + NamespaceCapabilities capabilities = new NamespaceCapabilities(); + capabilities.setFromLimits( + NamespaceInfo.Limits.newBuilder() + .setBlobSizeLimitError(blob) + .setMemoSizeLimitError(memo) + .build()); + return capabilities; + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowWorkerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowWorkerTest.java index efb1f039a7..ff231a4c46 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowWorkerTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowWorkerTest.java @@ -20,6 +20,8 @@ import com.uber.m3.tally.RootScopeBuilder; import com.uber.m3.tally.Scope; import com.uber.m3.util.ImmutableMap; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; import io.temporal.api.command.v1.Command; import io.temporal.api.command.v1.CompleteWorkflowExecutionCommandAttributes; import io.temporal.api.command.v1.ContinueAsNewWorkflowExecutionCommandAttributes; @@ -43,6 +45,11 @@ import io.temporal.common.reporter.TestStatsReporter; import io.temporal.internal.common.InternalUtils; import io.temporal.internal.concurrent.structured.CancelSource; +import io.temporal.internal.payload.limits.LimitClass; +import io.temporal.internal.payload.limits.LimitSeverity; +import io.temporal.internal.payload.limits.PayloadErrorLimits; +import io.temporal.internal.payload.limits.PayloadLimitViolation; +import io.temporal.internal.payload.limits.PayloadLimitViolationException; import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.internal.payload.storage.StorageOperationMetrics; import io.temporal.internal.payload.storage.TestStorageDriver; @@ -52,6 +59,7 @@ import io.temporal.payload.storage.ExternalStorage; import io.temporal.payload.storage.StorageDriverTargetInfo; import io.temporal.payload.storage.StorageDriverWorkflowInfo; +import io.temporal.serviceclient.MetricsTag; import io.temporal.serviceclient.WorkflowServiceStubs; import io.temporal.testUtils.Eventually; import io.temporal.testUtils.HistoryUtils; @@ -717,6 +725,173 @@ public void anOversizedCompletionStillFailsWithoutExternalStorage() throws Excep sent.getValue().getCause()); } + @Test + public void aCompletionOverThePayloadErrorLimitFailsTheTaskRetryably() throws Exception { + NamespaceCapabilities capabilities = payloadErrorLimits(); + ArgumentCaptor sent = + ArgumentCaptor.forClass(RespondWorkflowTaskFailedRequest.class); + + ListAppender logs = captureWorkerLogs(); + try { + runOneTask( + null, + capabilities, + smallCompletion(), + CancellationToken.none(), + 1, + blockingStub -> + when(blockingStub.respondWorkflowTaskCompleted( + any(RespondWorkflowTaskCompletedRequest.class))) + .thenThrow(payloadLimitViolation()), + blockingStub -> verify(blockingStub).respondWorkflowTaskFailed(sent.capture())); + } finally { + stopCapturingWorkerLogs(logs); + } + assertNoWarnings(logs); + + assertEquals( + WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_PAYLOADS_TOO_LARGE, + sent.getValue().getCause()); + Failure failure = sent.getValue().getFailure(); + assertEquals(PAYLOAD_LIMIT_ERROR_MESSAGE, failure.getMessage()); + assertEquals("PayloadsTooLarge", failure.getApplicationFailureInfo().getType()); + assertFalse(failure.getApplicationFailureInfo().getNonRetryable()); + reporter.assertCounter( + MetricsType.WORKFLOW_TASK_EXECUTION_FAILURE_COUNTER, + ImmutableMap.of( + "worker_type", + "WorkflowWorker", + "workflow_type", + WORKFLOW_TYPE, + MetricsTag.TASK_FAILURE_TYPE, + MetricsTag.TASK_FAILURE_VALUE_PAYLOADS_TOO_LARGE), + 1); + } + + @Test + public void aLaterAttemptOverThePayloadErrorLimitIsLeftToTimeOut() throws Exception { + ListAppender logs = captureWorkerLogs(); + try { + runOneTask( + null, + payloadErrorLimits(), + smallCompletion(), + CancellationToken.none(), + 2, + blockingStub -> + when(blockingStub.respondWorkflowTaskCompleted( + any(RespondWorkflowTaskCompletedRequest.class))) + .thenThrow(payloadLimitViolation()), + blockingStub -> + verify(blockingStub, never()) + .respondWorkflowTaskFailed(any(RespondWorkflowTaskFailedRequest.class))); + } finally { + stopCapturingWorkerLogs(logs); + } + assertNoWarnings(logs); + reporter.assertCounter( + MetricsType.WORKFLOW_TASK_EXECUTION_FAILURE_COUNTER, + ImmutableMap.of( + "worker_type", + "WorkflowWorker", + "workflow_type", + WORKFLOW_TYPE, + MetricsTag.TASK_FAILURE_TYPE, + MetricsTag.TASK_FAILURE_VALUE_PAYLOADS_TOO_LARGE), + 1); + } + + @Test + public void completionsCarryTheNamespacePayloadErrorLimits() throws Exception { + NamespaceCapabilities capabilities = payloadErrorLimits(); + runOneTask( + null, + capabilities, + smallCompletion(), + CancellationToken.none(), + 1, + blockingStub -> {}, + blockingStub -> + verify(blockingStub) + .withOption( + PayloadErrorLimits.CALL_OPTIONS_KEY, capabilities.getPayloadErrorLimits())); + } + + @Test + public void aViolationWhileReportingAFailureIsNotConverted() throws Exception { + // Only completions are rejected for payload limits; a workflow's own failure report must not + // be turned into a PAYLOADS_TOO_LARGE failure. + RespondWorkflowTaskFailedRequest taskFailed = + RespondWorkflowTaskFailedRequest.newBuilder() + .setCause( + WorkflowTaskFailedCause + .WORKFLOW_TASK_FAILED_CAUSE_WORKFLOW_WORKER_UNHANDLED_FAILURE) + .build(); + ArgumentCaptor sent = + ArgumentCaptor.forClass(RespondWorkflowTaskFailedRequest.class); + runOneTask( + null, + payloadErrorLimits(), + new WorkflowTaskHandler.Result( + WORKFLOW_TYPE, null, taskFailed, null, null, false, null, null), + CancellationToken.none(), + 1, + blockingStub -> {}, + blockingStub -> verify(blockingStub).respondWorkflowTaskFailed(sent.capture())); + assertEquals( + WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_WORKFLOW_WORKER_UNHANDLED_FAILURE, + sent.getValue().getCause()); + } + + // The payload limit validator logs the violation, so the worker must not log it again. + private static void assertNoWarnings(ListAppender logs) { + for (ILoggingEvent event : logs.list) { + assertFalse( + event.getLevel() + ": " + event.getFormattedMessage(), + event.getLevel().isGreaterOrEqual(Level.WARN)); + } + } + + private static final String PAYLOAD_LIMIT_ERROR_MESSAGE = + "[TMPRL1103] Attempted to upload payloads with size that exceeded the error limit."; + + private static NamespaceCapabilities payloadErrorLimits() { + NamespaceCapabilities capabilities = new NamespaceCapabilities(); + capabilities.setFromLimits( + NamespaceInfo.Limits.newBuilder() + .setBlobSizeLimitError(100) + .setMemoSizeLimitError(100) + .build()); + return capabilities; + } + + /** Returns what the payload limits interceptor throws for a completion over an error limit. */ + private static StatusRuntimeException payloadLimitViolation() { + PayloadLimitViolation violation = + new PayloadLimitViolation( + "commands[0].complete_workflow_execution_command_attributes.result", + LimitClass.BLOB, + LimitSeverity.ERROR, + 1000, + 100); + return Status.INVALID_ARGUMENT + .withDescription(violation.getMessage()) + .withCause(new PayloadLimitViolationException(violation)) + .asRuntimeException(); + } + + private static WorkflowTaskHandler.Result smallCompletion() { + RespondWorkflowTaskCompletedRequest taskCompleted = + RespondWorkflowTaskCompletedRequest.newBuilder() + .addCommands( + Command.newBuilder() + .setCompleteWorkflowExecutionCommandAttributes( + CompleteWorkflowExecutionCommandAttributes.getDefaultInstance())) + .build(); + return new WorkflowTaskHandler.Result( + WORKFLOW_TYPE, taskCompleted, null, null, null, false, null, null); + } + private static final int ONE_MEGABYTE = 1024 * 1024; private static NamespaceCapabilities completionSizeLimit(long limitBytes) { @@ -783,6 +958,28 @@ private void runOneTask( CancellationToken storageCancellation, java.util.function.Consumer verification) throws Exception { + runOneTask( + driver, + namespaceCapabilities, + handlerResult, + storageCancellation, + 1, + blockingStub -> {}, + verification); + } + + /** + * @param stubSetup runs after the default stubbing, so it can override it. + */ + private void runOneTask( + @Nullable TestStorageDriver driver, + NamespaceCapabilities namespaceCapabilities, + WorkflowTaskHandler.Result handlerResult, + CancellationToken storageCancellation, + int attempt, + java.util.function.Consumer stubSetup, + java.util.function.Consumer verification) + throws Exception { WorkflowServiceStubs client = mock(WorkflowServiceStubs.class); when(client.getServerCapabilities()) .thenReturn(() -> GetSystemInfoResponse.Capabilities.newBuilder().build()); @@ -839,10 +1036,12 @@ private void runOneTask( when(blockingStub.withOption(any(), any())).thenReturn(blockingStub); when(blockingStub.respondWorkflowTaskCompleted(any(RespondWorkflowTaskCompletedRequest.class))) .thenReturn(RespondWorkflowTaskCompletedResponse.getDefaultInstance()); + stubSetup.accept(blockingStub); PollWorkflowTaskQueueResponse pollResponse = PollWorkflowTaskQueueResponse.newBuilder() .setTaskToken(ByteString.copyFrom("token", UTF_8)) + .setAttempt(attempt) .setWorkflowExecution( WorkflowExecution.newBuilder().setWorkflowId(WORKFLOW_ID).setRunId(RUN_ID).build()) .setWorkflowType(WorkflowType.newBuilder().setName(WORKFLOW_TYPE).build()) diff --git a/temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java b/temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java index 1dd61df1d2..955ef7e31a 100644 --- a/temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java +++ b/temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java @@ -27,6 +27,20 @@ private void verifyBuild(WorkerOptions options) { assertNotNull(options.getPreferredVersionProvider()); } + @Test + public void payloadErrorLimitIsEnforcedByDefaultAndCanBeDisabled() { + assertFalse(WorkerOptions.newBuilder().build().isPayloadErrorLimitDisabled()); + + WorkerOptions disabled = WorkerOptions.newBuilder().setDisablePayloadErrorLimit(true).build(); + assertTrue(disabled.isPayloadErrorLimitDisabled()); + assertTrue(WorkerOptions.newBuilder(disabled).build().isPayloadErrorLimitDisabled()); + assertTrue( + WorkerOptions.newBuilder(disabled) + .validateAndBuildWithDefaults() + .isPayloadErrorLimitDisabled()); + assertNotEquals(WorkerOptions.newBuilder().build(), disabled); + } + @Test public void verifyWorkerOptionsEquality() { WorkerOptions w1 = WorkerOptions.newBuilder().build(); diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/GrpcMessageTooLargeTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/GrpcMessageTooLargeTest.java index 20b4ec3368..18218b7013 100644 --- a/temporal-sdk/src/test/java/io/temporal/workflow/GrpcMessageTooLargeTest.java +++ b/temporal-sdk/src/test/java/io/temporal/workflow/GrpcMessageTooLargeTest.java @@ -15,6 +15,7 @@ import io.temporal.internal.worker.PollerOptions; import io.temporal.testUtils.LoggerUtils; import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.worker.WorkerOptions; import io.temporal.workflow.shared.TestActivities; import java.time.Duration; import java.util.List; @@ -36,16 +37,25 @@ public class GrpcMessageTooLargeTest { VERY_LARGE_DATA = argBuilder.toString(); } + // These workers send payloads over the namespace's error limit, which the worker would otherwise + // reject before the gRPC message size limit is reached. + private static final WorkerOptions PAYLOAD_ERROR_LIMIT_DISABLED = + WorkerOptions.newBuilder().setDisablePayloadErrorLimit(true).build(); + @Rule public SDKTestWorkflowRule activityStartWorkflowRule = SDKTestWorkflowRule.newBuilder() .setWorkflowTypes(ActivityStartWorkflowImpl.class) .setActivityImplementations(new TestActivityImpl()) + .setWorkerOptions(PAYLOAD_ERROR_LIMIT_DISABLED) .build(); @Rule public SDKTestWorkflowRule failureWorkflowRule = - SDKTestWorkflowRule.newBuilder().setWorkflowTypes(FailureWorkflowImpl.class).build(); + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(FailureWorkflowImpl.class) + .setWorkerOptions(PAYLOAD_ERROR_LIMIT_DISABLED) + .build(); @Rule public SDKTestWorkflowRule querySuccessWorkflowRule = diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/PayloadLimitsDisabledTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/PayloadLimitsDisabledTest.java new file mode 100644 index 0000000000..690636b33b --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/PayloadLimitsDisabledTest.java @@ -0,0 +1,40 @@ +package io.temporal.workflow; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assume.assumeTrue; + +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.worker.WorkerOptions; +import io.temporal.workflow.PayloadLimitsTest.ActivityFailureTypeWorkflow; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; + +/** + * With enforcement disabled, the worker sends oversized completions and the server rejects them. + */ +public class PayloadLimitsDisabledTest { + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(PayloadLimitsTest.ActivityFailureTypeWorkflowImpl.class) + .setActivityImplementations(new PayloadLimitsTest.OversizedActivitiesImpl()) + .setWorkerOptions(WorkerOptions.newBuilder().setDisablePayloadErrorLimit(true).build()) + .build(); + + @Before + public void requireRealServer() { + assumeTrue( + "Needs a server that reports payload error limits", SDKTestWorkflowRule.useExternalService); + } + + @Test + public void anOversizedActivityResultIsLeftToTheServer() { + ActivityFailureTypeWorkflow workflow = + testWorkflowRule.newWorkflowStubTimeoutOptions(ActivityFailureTypeWorkflow.class); + // The server fails the activity itself, with a non-retryable ServerFailure instead of the + // worker's retryable PayloadsTooLarge. + assertEquals("ServerFailure", workflow.execute(false)); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/PayloadLimitsTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/PayloadLimitsTest.java new file mode 100644 index 0000000000..9cea42d208 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/PayloadLimitsTest.java @@ -0,0 +1,208 @@ +package io.temporal.workflow; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; +import static org.junit.Assume.assumeTrue; + +import io.temporal.activity.Activity; +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import io.temporal.activity.ActivityOptions; +import io.temporal.api.enums.v1.EventType; +import io.temporal.api.enums.v1.WorkflowExecutionStatus; +import io.temporal.api.enums.v1.WorkflowTaskFailedCause; +import io.temporal.api.history.v1.HistoryEvent; +import io.temporal.api.history.v1.WorkflowTaskFailedEventAttributes; +import io.temporal.api.workflowservice.v1.DescribeWorkflowExecutionRequest; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowOptions; +import io.temporal.client.WorkflowServiceException; +import io.temporal.client.WorkflowStub; +import io.temporal.common.RetryOptions; +import io.temporal.failure.ActivityFailure; +import io.temporal.failure.ApplicationFailure; +import io.temporal.internal.payload.limits.PayloadLimitViolationException; +import io.temporal.testUtils.Eventually; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import java.time.Duration; +import java.util.Arrays; +import java.util.List; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; + +/** + * Tests worker enforcement of the namespace's payload error limits. They need a real server: the + * in-memory test server reports no limits, so the worker would enforce none. + */ +public class PayloadLimitsTest { + + /** This size is above the server's default 2 MiB blob error limit and below gRPC's 4 MiB cap. */ + static final String OVERSIZED = repeat('x', 3 * 1024 * 1024); + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes( + OversizedResultWorkflowImpl.class, ActivityFailureTypeWorkflowImpl.class) + .setActivityImplementations(new OversizedActivitiesImpl()) + .build(); + + @Before + public void requireRealServer() { + assumeTrue( + "Needs a server that reports payload error limits", SDKTestWorkflowRule.useExternalService); + } + + @Test + public void anOversizedWorkflowResultFailsTheWorkflowTaskRetryably() { + OversizedResultWorkflow workflow = + testWorkflowRule.newWorkflowStubTimeoutOptions(OversizedResultWorkflow.class); + WorkflowClient.start(workflow::execute); + String workflowId = WorkflowStub.fromTyped(workflow).getExecution().getWorkflowId(); + + WorkflowTaskFailedEventAttributes failed = + Eventually.assertEventually( + Duration.ofSeconds(20), + () -> { + List events = + testWorkflowRule.getHistoryEvents( + workflowId, EventType.EVENT_TYPE_WORKFLOW_TASK_FAILED); + assertFalse(events.isEmpty()); + return events.get(0).getWorkflowTaskFailedEventAttributes(); + }); + + assertEquals( + WorkflowTaskFailedCause.WORKFLOW_TASK_FAILED_CAUSE_PAYLOADS_TOO_LARGE, failed.getCause()); + assertEquals("PayloadsTooLarge", failed.getFailure().getApplicationFailureInfo().getType()); + assertFalse(failed.getFailure().getApplicationFailureInfo().getNonRetryable()); + // The workflow stays running, so that a corrected version can be deployed. + assertEquals( + WorkflowExecutionStatus.WORKFLOW_EXECUTION_STATUS_RUNNING, + testWorkflowRule + .getWorkflowClient() + .getWorkflowServiceStubs() + .blockingStub() + .describeWorkflowExecution( + DescribeWorkflowExecutionRequest.newBuilder() + .setNamespace(testWorkflowRule.getWorkflowClient().getOptions().getNamespace()) + .setExecution(WorkflowStub.fromTyped(workflow).getExecution()) + .build()) + .getWorkflowExecutionInfo() + .getStatus()); + WorkflowStub.fromTyped(workflow).terminate("done"); + } + + @Test + public void anOversizedActivityResultFailsTheActivityRetryably() { + ActivityFailureTypeWorkflow workflow = + testWorkflowRule.newWorkflowStubTimeoutOptions(ActivityFailureTypeWorkflow.class); + assertEquals("PayloadsTooLarge retryable", workflow.execute(false)); + } + + @Test + public void anOversizedHeartbeatFailsTheActivityRetryably() { + ActivityFailureTypeWorkflow workflow = + testWorkflowRule.newWorkflowStubTimeoutOptions(ActivityFailureTypeWorkflow.class); + assertEquals("PayloadsTooLarge retryable", workflow.execute(true)); + } + + @Test + public void clientCallsOverTheErrorLimitAreSentNotRejectedLocally() { + WorkflowStub workflow = + testWorkflowRule + .getWorkflowClient() + .newUntypedWorkflowStub( + "OversizedResultWorkflow", + WorkflowOptions.newBuilder() + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setWorkflowRunTimeout(Duration.ofSeconds(10)) + .build()); + // The client only warns; the server is the one that rejects the oversized input. + WorkflowServiceException e = + assertThrows(WorkflowServiceException.class, () -> workflow.start(OVERSIZED)); + assertFalse(PayloadLimitViolationException.find(e).isPresent()); + } + + @WorkflowInterface + public interface OversizedResultWorkflow { + @WorkflowMethod + String execute(); + } + + public static class OversizedResultWorkflowImpl implements OversizedResultWorkflow { + @Override + public String execute() { + return OVERSIZED; + } + } + + @WorkflowInterface + public interface ActivityFailureTypeWorkflow { + /** Returns the activity failure's type and whether it was retryable. */ + @WorkflowMethod + String execute(boolean heartbeat); + } + + public static class ActivityFailureTypeWorkflowImpl implements ActivityFailureTypeWorkflow { + private final OversizedActivities activities = + Workflow.newActivityStub( + OversizedActivities.class, + ActivityOptions.newBuilder() + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .setHeartbeatTimeout(Duration.ofSeconds(5)) + .setRetryOptions(RetryOptions.newBuilder().setMaximumAttempts(1).build()) + .build()); + + @Override + public String execute(boolean heartbeat) { + try { + if (heartbeat) { + activities.heartbeatOversized(); + } else { + activities.returnOversized(); + } + return "completed"; + } catch (ActivityFailure e) { + return describe(e.getCause()); + } + } + } + + @ActivityInterface + public interface OversizedActivities { + @ActivityMethod + String returnOversized(); + + @ActivityMethod + void heartbeatOversized(); + } + + public static class OversizedActivitiesImpl implements OversizedActivities { + @Override + public String returnOversized() { + return OVERSIZED; + } + + @Override + public void heartbeatOversized() { + Activity.getExecutionContext().heartbeat(OVERSIZED); + } + } + + static String describe(Throwable failure) { + if (failure instanceof ApplicationFailure) { + ApplicationFailure applicationFailure = (ApplicationFailure) failure; + return applicationFailure.getType() + + (applicationFailure.isNonRetryable() ? " non-retryable" : " retryable"); + } + return failure == null ? "null" : failure.getClass().getSimpleName(); + } + + private static String repeat(char c, int count) { + char[] chars = new char[count]; + Arrays.fill(chars, c); + return new String(chars); + } +} diff --git a/temporal-serviceclient/build.gradle b/temporal-serviceclient/build.gradle index ec63fa64bf..b698e1dbe6 100644 --- a/temporal-serviceclient/build.gradle +++ b/temporal-serviceclient/build.gradle @@ -35,7 +35,7 @@ dependencies { testRuntimeOnly "com.google.protobuf:protobuf-java-util:$protoVersionEdge" testRuntimeOnly "io.grpc:grpc-bom:$grpcVersionEdge" } - testRuntimeOnly "ch.qos.logback:logback-classic:${logbackVersion}" + testImplementation "ch.qos.logback:logback-classic:${logbackVersion}" } task initSubmodules(type: Exec) { diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/LimitClass.java b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/LimitClass.java index 515b10b2ad..9b1543aa0e 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/LimitClass.java +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/LimitClass.java @@ -1,7 +1,7 @@ package io.temporal.internal.payload.limits; /** Which server-enforced size limit a payload field is subject to. */ -enum LimitClass { +public enum LimitClass { /** Subject to the blob (payload) size limit. */ BLOB, /** Subject to the memo size limit. */ diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/LimitSeverity.java b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/LimitSeverity.java index f97eb421f1..5bd9f295d4 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/LimitSeverity.java +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/LimitSeverity.java @@ -1,7 +1,7 @@ package io.temporal.internal.payload.limits; /** Whether a violation exceeded the warning threshold or the error threshold. */ -enum LimitSeverity { +public enum LimitSeverity { /** Exceeded the warning threshold. */ WARNING, /** Exceeded the error threshold. */ diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadErrorLimits.java b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadErrorLimits.java new file mode 100644 index 0000000000..1add4a9bf4 --- /dev/null +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadErrorLimits.java @@ -0,0 +1,38 @@ +package io.temporal.internal.payload.limits; + +import io.grpc.CallOptions; + +/** + * Holds the namespace's payload and memo error limits, in bytes, which are attached to a single + * call through {@link #CALL_OPTIONS_KEY}. Only workers attach them, so calls made by clients are + * checked against the warning thresholds only. A {@code 0} limit disables error enforcement for + * that class. + */ +public final class PayloadErrorLimits { + /** Enforces no error limits. Calls without the option behave as if they carried this. */ + public static final PayloadErrorLimits NONE = new PayloadErrorLimits(0, 0); + + public static final CallOptions.Key CALL_OPTIONS_KEY = + CallOptions.Key.createWithDefault("temporal-payload-error-limits", NONE); + + private final long blob; + private final long memo; + + public PayloadErrorLimits(long blob, long memo) { + this.blob = blob; + this.memo = memo; + } + + public long getBlob() { + return blob; + } + + public long getMemo() { + return memo; + } + + @Override + public String toString() { + return "PayloadErrorLimits{blob=" + blob + ", memo=" + memo + '}'; + } +} diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitViolation.java b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitViolation.java index 6c8ddf6728..b9fd0c277a 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitViolation.java +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitViolation.java @@ -1,14 +1,14 @@ package io.temporal.internal.payload.limits; /** A payload field whose size exceeded one of its configured thresholds (warning or error). */ -final class PayloadLimitViolation { +public final class PayloadLimitViolation { private final String path; private final LimitClass limitClass; private final LimitSeverity severity; private final long size; private final long limit; - PayloadLimitViolation( + public PayloadLimitViolation( String path, LimitClass limitClass, LimitSeverity severity, long size, long limit) { this.path = path; this.limitClass = limitClass; @@ -21,32 +21,32 @@ final class PayloadLimitViolation { * Path of proto field names from the root message (e.g. {@code * commands[2].schedule_activity_task_command_attributes.input}). */ - String getPath() { + public String getPath() { return path; } - LimitClass getLimitClass() { + public LimitClass getLimitClass() { return limitClass; } - LimitSeverity getSeverity() { + public LimitSeverity getSeverity() { return severity; } /** The field's measured size in bytes. */ - long getSize() { + public long getSize() { return size; } /** * The threshold that was exceeded (warning threshold for warnings, error threshold for errors). */ - long getLimit() { + public long getLimit() { return limit; } /** The user-facing {@code [TMPRL1103]} message. */ - String getMessage() { + public String getMessage() { String limitClassName = limitClass == LimitClass.BLOB ? "payloads" : "memo"; String limitKind = severity == LimitSeverity.WARNING ? "warning" : "error"; return "[TMPRL1103] Attempted to upload " diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitViolationException.java b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitViolationException.java new file mode 100644 index 0000000000..42df27b275 --- /dev/null +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitViolationException.java @@ -0,0 +1,40 @@ +package io.temporal.internal.payload.limits; + +import java.util.Optional; +import javax.annotation.Nullable; + +/** + * Signals that a call failed locally because its request exceeds an error limit; it is the cause of + * the call's {@code INVALID_ARGUMENT} status. Workers look for it with {@link #find(Throwable)} to + * replace the rejected completion with a task failure. + */ +public final class PayloadLimitViolationException extends RuntimeException { + /** A worker fails a task with an {@code ApplicationFailure} of this type on a violation. */ + public static final String FAILURE_TYPE = "PayloadsTooLarge"; + + private static final int MAX_CAUSE_DEPTH = 32; + + private final PayloadLimitViolation violation; + + public PayloadLimitViolationException(PayloadLimitViolation violation) { + // The violation is reported through the task failure, so a stack trace adds nothing. + super(violation.getMessage(), null, false, false); + this.violation = violation; + } + + public PayloadLimitViolation getViolation() { + return violation; + } + + /** Returns the violation that caused {@code throwable}, searching its cause chain. */ + public static Optional find(@Nullable Throwable throwable) { + Throwable t = throwable; + for (int depth = 0; t != null && depth < MAX_CAUSE_DEPTH; depth++) { + if (t instanceof PayloadLimitViolationException) { + return Optional.of(((PayloadLimitViolationException) t).violation); + } + t = t.getCause(); + } + return Optional.empty(); + } +} diff --git a/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitsInterceptor.java b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitsInterceptor.java new file mode 100644 index 0000000000..d396feaf28 --- /dev/null +++ b/temporal-serviceclient/src/main/java/io/temporal/internal/payload/limits/PayloadLimitsInterceptor.java @@ -0,0 +1,183 @@ +package io.temporal.internal.payload.limits; + +import com.google.protobuf.Message; +import io.grpc.Attributes; +import io.grpc.CallOptions; +import io.grpc.Channel; +import io.grpc.ClientCall; +import io.grpc.ClientInterceptor; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.Status; +import java.util.Optional; +import javax.annotation.Nullable; + +/** + * Checks the payload and memo fields of every outbound request against the server's size limits. + * + *

Fields over a warning threshold are logged and the request is sent. When the call carries + * {@link PayloadErrorLimits} and a field exceeds them, the request is not sent: the call fails with + * {@code INVALID_ARGUMENT}, caused by a {@link PayloadLimitViolationException}. + * + *

Must run after any interceptor that can change the request, so that it checks what is actually + * sent. + */ +public final class PayloadLimitsInterceptor implements ClientInterceptor { + private final long payloadsWarnSize; + private final long memoWarnSize; + + public PayloadLimitsInterceptor(long payloadsWarnSize, long memoWarnSize) { + this.payloadsWarnSize = payloadsWarnSize; + this.memoWarnSize = memoWarnSize; + } + + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + // Every payload-bearing Temporal RPC is unary; deferring a streaming call would change its flow + // control. + if (method.getType() != MethodDescriptor.MethodType.UNARY) { + return next.newCall(method, callOptions); + } + PayloadErrorLimits errorLimits = callOptions.getOption(PayloadErrorLimits.CALL_OPTIONS_KEY); + PayloadLimits limits = + new PayloadLimits( + payloadsWarnSize, errorLimits.getBlob(), memoWarnSize, errorLimits.getMemo()); + if (limits.warn(LimitClass.BLOB) == 0 + && limits.error(LimitClass.BLOB) == 0 + && limits.warn(LimitClass.MEMO) == 0 + && limits.error(LimitClass.MEMO) == 0) { + return next.newCall(method, callOptions); + } + return new ValidatingCall<>(next, method, callOptions, limits); + } + + /** + * Holds back creating and starting the delegate until the request is seen. A rejected request + * then never opens a stream to the server, and the interceptors inside this one, such as the one + * that counts requests, never see it. + */ + private static final class ValidatingCall extends ClientCall { + private final Channel next; + private final MethodDescriptor method; + private final CallOptions callOptions; + private final PayloadLimits limits; + + private ClientCall delegate; + + private Listener listener; + private Metadata headers; + private int pendingRequests; + private Boolean messageCompression; + private boolean started; + private boolean rejected; + + ValidatingCall( + Channel next, + MethodDescriptor method, + CallOptions callOptions, + PayloadLimits limits) { + this.next = next; + this.method = method; + this.callOptions = callOptions; + this.limits = limits; + } + + @Override + public void start(Listener responseListener, Metadata headers) { + this.listener = responseListener; + this.headers = headers; + } + + @Override + public void request(int numMessages) { + if (started) { + delegate.request(numMessages); + } else { + pendingRequests += numMessages; + } + } + + @Override + public void sendMessage(ReqT message) { + if (rejected) { + return; + } + if (message instanceof Message) { + Optional violation = + PayloadLimitValidator.validate((Message) message, limits); + if (violation.isPresent()) { + rejected = true; + listener.onClose( + Status.INVALID_ARGUMENT + .withDescription(violation.get().getMessage()) + .withCause(new PayloadLimitViolationException(violation.get())), + new Metadata()); + return; + } + } + startDelegate(); + delegate.sendMessage(message); + } + + @Override + public void halfClose() { + if (rejected) { + return; + } + startDelegate(); + delegate.halfClose(); + } + + @Override + public void cancel(@Nullable String message, @Nullable Throwable cause) { + if (rejected) { + return; + } + if (started) { + delegate.cancel(message, cause); + return; + } + // Nothing was sent, so close the listener here rather than create a call only to cancel it. + rejected = true; + if (listener != null) { + listener.onClose( + Status.CANCELLED.withDescription(message).withCause(cause), new Metadata()); + } + } + + @Override + public boolean isReady() { + return started && delegate.isReady(); + } + + @Override + public void setMessageCompression(boolean enabled) { + if (started) { + delegate.setMessageCompression(enabled); + } else { + messageCompression = enabled; + } + } + + @Override + public Attributes getAttributes() { + return started ? delegate.getAttributes() : Attributes.EMPTY; + } + + private void startDelegate() { + if (started) { + return; + } + started = true; + delegate = next.newCall(method, callOptions); + delegate.start(listener, headers); + if (messageCompression != null) { + delegate.setMessageCompression(messageCompression); + } + if (pendingRequests > 0) { + delegate.request(pendingRequests); + } + } + } +} diff --git a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/ChannelManager.java b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/ChannelManager.java index 4d32633d36..44f511d1c1 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/ChannelManager.java +++ b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/ChannelManager.java @@ -8,6 +8,7 @@ import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder; import io.grpc.stub.MetadataUtils; import io.temporal.api.workflowservice.v1.GetSystemInfoResponse.Capabilities; +import io.temporal.internal.payload.limits.PayloadLimitsInterceptor; import io.temporal.internal.retryer.GrpcRetryer; import io.temporal.internal.retryer.GrpcRetryer.GrpcRetryerOptions; import java.time.Duration; @@ -130,6 +131,19 @@ private Channel applyTailStandardInterceptors(Channel channel) { channel = ClientInterceptors.intercept(channel, metricsInterceptor); + // The payload limits interceptor sits inside the custom interceptors, so it checks the request + // as finally sent, and outside the metrics interceptor, so a request it rejects locally is not + // counted as an RPC. + if (options instanceof WorkflowServiceStubsOptions) { + PayloadLimitsOptions payloadLimits = + ((WorkflowServiceStubsOptions) options).getPayloadLimits(); + channel = + ClientInterceptors.intercept( + channel, + new PayloadLimitsInterceptor( + payloadLimits.getPayloadsWarnSize(), payloadLimits.getMemoWarnSize())); + } + // if this interceptor is enabled, it should be added first or in front of any requests // modifying interceptors // to have the access to fully formed requests diff --git a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/MetricsTag.java b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/MetricsTag.java index 69b64ecc6b..1740bb494d 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/MetricsTag.java +++ b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/MetricsTag.java @@ -27,6 +27,7 @@ public class MetricsTag { public static final String TASK_FAILURE_VALUE_NON_DETERMINISM_ERROR = "NonDeterminismError"; public static final String TASK_FAILURE_VALUE_GRPC_MESSAGE_TOO_LARGE = "GrpcMessageTooLarge"; public static final String TASK_FAILURE_VALUE_REQUEST_TOO_LARGE = "RequestTooLarge"; + public static final String TASK_FAILURE_VALUE_PAYLOADS_TOO_LARGE = "PayloadsTooLarge"; public static final String TASK_FAILURE_VALUE_WORKFLOW_ERROR = "WorkflowError"; public static final String TASK_FAILURE_VALUE_ACTIVITY_ERROR = "ActivityError"; public static final String TASK_FAILURE_VALUE_OPERATION_FAILED = "operation_failed"; diff --git a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/PayloadLimitsOptions.java b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/PayloadLimitsOptions.java new file mode 100644 index 0000000000..b58c1645b7 --- /dev/null +++ b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/PayloadLimitsOptions.java @@ -0,0 +1,122 @@ +package io.temporal.serviceclient; + +import com.google.common.base.MoreObjects; +import com.google.common.base.Preconditions; +import java.util.Objects; + +/** + * Configures warning thresholds for the size of outbound payload and memo fields. A field over its + * threshold is logged with a {@code [TMPRL1103]} warning but still sent to the server. + * + *

Workers additionally enforce the namespace's error limits, failing a workflow or activity task + * whose completion exceeds them instead of sending it. See {@code + * WorkerOptions.Builder#setDisablePayloadErrorLimit}. + * + *

WARNING: The payload limit warning options are experimental and may change in the future. + */ +public final class PayloadLimitsOptions { + + private static final long DEFAULT_PAYLOADS_WARN_SIZE = 512 * 1024; + private static final long DEFAULT_MEMO_WARN_SIZE = 2 * 1024; + + private static final PayloadLimitsOptions DEFAULT_INSTANCE = newBuilder().build(); + + public static Builder newBuilder() { + return new Builder(); + } + + public static Builder newBuilder(PayloadLimitsOptions options) { + return new Builder(options); + } + + public static PayloadLimitsOptions getDefaultInstance() { + return DEFAULT_INSTANCE; + } + + private final long payloadsWarnSize; + private final long memoWarnSize; + + private PayloadLimitsOptions(long payloadsWarnSize, long memoWarnSize) { + this.payloadsWarnSize = payloadsWarnSize; + this.memoWarnSize = memoWarnSize; + } + + /** + * @return the warning threshold, in bytes, for the size of an outbound payload-bearing field; 0 + * means the warning is disabled. + */ + public long getPayloadsWarnSize() { + return payloadsWarnSize; + } + + /** + * @return the warning threshold, in bytes, for the size of an outbound memo; 0 means the warning + * is disabled. + */ + public long getMemoWarnSize() { + return memoWarnSize; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + PayloadLimitsOptions that = (PayloadLimitsOptions) o; + return payloadsWarnSize == that.payloadsWarnSize && memoWarnSize == that.memoWarnSize; + } + + @Override + public int hashCode() { + return Objects.hash(payloadsWarnSize, memoWarnSize); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("payloadsWarnSize", payloadsWarnSize) + .add("memoWarnSize", memoWarnSize) + .toString(); + } + + public static final class Builder { + private long payloadsWarnSize = DEFAULT_PAYLOADS_WARN_SIZE; + private long memoWarnSize = DEFAULT_MEMO_WARN_SIZE; + + private Builder() {} + + private Builder(PayloadLimitsOptions options) { + if (options == null) { + return; + } + this.payloadsWarnSize = options.payloadsWarnSize; + this.memoWarnSize = options.memoWarnSize; + } + + /** + * Sets the 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; set to 0 + * to disable. + */ + public Builder setPayloadsWarnSize(long payloadsWarnSize) { + this.payloadsWarnSize = payloadsWarnSize; + return this; + } + + /** + * Sets the warning threshold, in bytes, for the size of an outbound memo. Over-threshold memos + * are logged but still sent to the server. Defaults to 2 KiB; set to 0 to disable. + */ + public Builder setMemoWarnSize(long memoWarnSize) { + this.memoWarnSize = memoWarnSize; + return this; + } + + public PayloadLimitsOptions build() { + Preconditions.checkArgument( + payloadsWarnSize >= 0, "payloadsWarnSize must not be negative: %s", payloadsWarnSize); + Preconditions.checkArgument( + memoWarnSize >= 0, "memoWarnSize must not be negative: %s", memoWarnSize); + return new PayloadLimitsOptions(payloadsWarnSize, memoWarnSize); + } + } +} diff --git a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/WorkflowServiceStubsOptions.java b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/WorkflowServiceStubsOptions.java index f22a2ac67e..09e094e747 100644 --- a/temporal-serviceclient/src/main/java/io/temporal/serviceclient/WorkflowServiceStubsOptions.java +++ b/temporal-serviceclient/src/main/java/io/temporal/serviceclient/WorkflowServiceStubsOptions.java @@ -40,6 +40,9 @@ public final class WorkflowServiceStubsOptions extends ServiceStubsOptions { /** Plugins for customizing service stubs configuration and connection */ private final WorkflowServiceStubsPlugin[] plugins; + /** Warning thresholds for outbound payload and memo sizes */ + private final PayloadLimitsOptions payloadLimits; + private static final WorkflowServiceStubsPlugin[] EMPTY_PLUGINS = new WorkflowServiceStubsPlugin[0]; @@ -61,13 +64,15 @@ private WorkflowServiceStubsOptions( Duration rpcLongPollTimeout, Duration rpcQueryTimeout, RpcRetryOptions rpcRetryOptions, - WorkflowServiceStubsPlugin[] plugins) { + WorkflowServiceStubsPlugin[] plugins, + PayloadLimitsOptions payloadLimits) { super(serviceStubsOptions); this.disableHealthCheck = disableHealthCheck; this.rpcLongPollTimeout = rpcLongPollTimeout; this.rpcQueryTimeout = rpcQueryTimeout; this.rpcRetryOptions = rpcRetryOptions; this.plugins = plugins; + this.payloadLimits = payloadLimits; } /** @@ -114,6 +119,17 @@ public WorkflowServiceStubsPlugin[] getPlugins() { return plugins; } + /** + * Returns the warning thresholds for outbound payload and memo sizes. + * + *

WARNING: The payload limit warning options are experimental and may change in the future. + * + * @return the payload limits options, never null. + */ + public PayloadLimitsOptions getPayloadLimits() { + return payloadLimits; + } + @Override public boolean equals(Object o) { if (this == o) return true; @@ -123,13 +139,19 @@ public boolean equals(Object o) { && Objects.equals(rpcLongPollTimeout, that.rpcLongPollTimeout) && Objects.equals(rpcQueryTimeout, that.rpcQueryTimeout) && Objects.equals(rpcRetryOptions, that.rpcRetryOptions) - && Arrays.equals(plugins, that.plugins); + && Arrays.equals(plugins, that.plugins) + && Objects.equals(payloadLimits, that.payloadLimits); } @Override public int hashCode() { int result = - Objects.hash(disableHealthCheck, rpcLongPollTimeout, rpcQueryTimeout, rpcRetryOptions); + Objects.hash( + disableHealthCheck, + rpcLongPollTimeout, + rpcQueryTimeout, + rpcRetryOptions, + payloadLimits); result = 31 * result + Arrays.hashCode(plugins); return result; } @@ -147,6 +169,8 @@ public String toString() { + rpcRetryOptions + ", plugins=" + Arrays.toString(plugins) + + ", payloadLimits=" + + payloadLimits + '}'; } @@ -157,6 +181,7 @@ public static class Builder extends ServiceStubsOptions.Builder { private Duration rpcQueryTimeout = DEFAULT_QUERY_RPC_TIMEOUT; private RpcRetryOptions rpcRetryOptions = DefaultStubServiceOperationRpcRetryOptions.INSTANCE; private WorkflowServiceStubsPlugin[] plugins; + private PayloadLimitsOptions payloadLimits = PayloadLimitsOptions.getDefaultInstance(); private Builder() {} @@ -168,6 +193,7 @@ private Builder(ServiceStubsOptions options) { this.rpcQueryTimeout = castedOptions.rpcQueryTimeout; this.rpcRetryOptions = castedOptions.rpcRetryOptions; this.plugins = castedOptions.plugins; + this.payloadLimits = castedOptions.payloadLimits; } } @@ -279,6 +305,21 @@ public Builder setPlugins(WorkflowServiceStubsPlugin... plugins) { return this; } + /** + * Sets the warning thresholds for outbound payload and memo sizes. Defaults to {@link + * PayloadLimitsOptions#getDefaultInstance()}: 512 KiB per payload-bearing field and 2 KiB per + * memo. + * + *

Every request sent through these stubs is checked: a field over its warning threshold is + * logged but still sent. Workers additionally enforce the namespace's error limits. + * + *

WARNING: The payload limit warning options are experimental and may change in the future. + */ + public Builder setPayloadLimits(PayloadLimitsOptions payloadLimits) { + this.payloadLimits = Objects.requireNonNull(payloadLimits); + return this; + } + /** * Sets the rpc timeout value for query calls. Default is 10 seconds. * @@ -302,7 +343,8 @@ public WorkflowServiceStubsOptions build() { this.rpcLongPollTimeout, this.rpcQueryTimeout, this.rpcRetryOptions, - this.plugins); + this.plugins, + this.payloadLimits); } /** @@ -326,7 +368,8 @@ public WorkflowServiceStubsOptions validateAndBuildWithDefaults() { this.rpcLongPollTimeout, this.rpcQueryTimeout, retryOptions, - this.plugins == null ? EMPTY_PLUGINS : this.plugins); + this.plugins == null ? EMPTY_PLUGINS : this.plugins, + this.payloadLimits); } } } diff --git a/temporal-serviceclient/src/test/java/io/temporal/serviceclient/PayloadLimitsInterceptorTest.java b/temporal-serviceclient/src/test/java/io/temporal/serviceclient/PayloadLimitsInterceptorTest.java new file mode 100644 index 0000000000..8ef325556c --- /dev/null +++ b/temporal-serviceclient/src/test/java/io/temporal/serviceclient/PayloadLimitsInterceptorTest.java @@ -0,0 +1,424 @@ +package io.temporal.serviceclient; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.argThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.LoggerContext; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; +import com.google.protobuf.ByteString; +import com.uber.m3.tally.Capabilities; +import com.uber.m3.tally.CapableOf; +import com.uber.m3.tally.RootScopeBuilder; +import com.uber.m3.tally.Scope; +import com.uber.m3.tally.StatsReporter; +import com.uber.m3.util.Duration; +import io.grpc.CallOptions; +import io.grpc.Channel; +import io.grpc.ClientCall; +import io.grpc.ClientInterceptor; +import io.grpc.ForwardingClientCall; +import io.grpc.ManagedChannel; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import io.grpc.inprocess.InProcessChannelBuilder; +import io.grpc.inprocess.InProcessServerBuilder; +import io.grpc.stub.StreamObserver; +import io.grpc.testing.GrpcCleanupRule; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.Payloads; +import io.temporal.api.workflowservice.v1.DescribeNamespaceRequest; +import io.temporal.api.workflowservice.v1.DescribeNamespaceResponse; +import io.temporal.api.workflowservice.v1.GetSystemInfoRequest; +import io.temporal.api.workflowservice.v1.GetSystemInfoResponse; +import io.temporal.api.workflowservice.v1.StartWorkflowExecutionRequest; +import io.temporal.api.workflowservice.v1.StartWorkflowExecutionResponse; +import io.temporal.api.workflowservice.v1.WorkflowServiceGrpc; +import io.temporal.api.workflowservice.v1.WorkflowServiceGrpc.WorkflowServiceImplBase; +import io.temporal.internal.payload.limits.PayloadErrorLimits; +import io.temporal.internal.payload.limits.PayloadLimitViolation; +import io.temporal.internal.payload.limits.PayloadLimitViolationException; +import io.temporal.internal.payload.limits.PayloadLimitsInterceptor; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.slf4j.LoggerFactory; + +public class PayloadLimitsInterceptorTest { + + private static final PayloadErrorLimits ERROR_LIMITS = new PayloadErrorLimits(100, 100); + + // The validator is package-private in another package, so its logger is named directly. + private static final String PAYLOAD_LIMIT_VALIDATOR_LOGGER = + "io.temporal.internal.payload.limits.PayloadLimitValidator"; + + @Rule public final GrpcCleanupRule grpcCleanupRule = new GrpcCleanupRule(); + + private final AtomicInteger startCount = new AtomicInteger(); + private final AtomicInteger describeCount = new AtomicInteger(); + + private final WorkflowServiceImplBase workflowImpl = + new WorkflowServiceImplBase() { + @Override + public void getSystemInfo( + GetSystemInfoRequest request, StreamObserver observer) { + observer.onNext(GetSystemInfoResponse.getDefaultInstance()); + observer.onCompleted(); + } + + @Override + public void startWorkflowExecution( + StartWorkflowExecutionRequest request, + StreamObserver observer) { + startCount.incrementAndGet(); + observer.onNext(StartWorkflowExecutionResponse.getDefaultInstance()); + observer.onCompleted(); + } + + @Override + public void describeNamespace( + DescribeNamespaceRequest request, StreamObserver observer) { + describeCount.incrementAndGet(); + observer.onNext(DescribeNamespaceResponse.getDefaultInstance()); + observer.onCompleted(); + } + }; + + private ManagedChannel channel; + + @Before + public void setUp() throws Exception { + String serverName = InProcessServerBuilder.generateName(); + grpcCleanupRule.register( + InProcessServerBuilder.forName(serverName) + .directExecutor() + .addService(workflowImpl) + .build() + .start()); + channel = + grpcCleanupRule.register( + InProcessChannelBuilder.forName(serverName).directExecutor().build()); + } + + private WorkflowServiceStubs stubs( + PayloadLimitsOptions payloadLimits, ClientInterceptor... interceptors) { + WorkflowServiceStubsOptions.Builder options = + WorkflowServiceStubsOptions.newBuilder() + .setChannel(channel) + .setPayloadLimits(payloadLimits); + if (interceptors.length > 0) { + options.setGrpcClientInterceptors(java.util.Arrays.asList(interceptors)); + } + WorkflowServiceStubs stubs = WorkflowServiceStubs.newServiceStubs(options.build()); + return stubs; + } + + private static StartWorkflowExecutionRequest startWithInput(int dataLen) { + return StartWorkflowExecutionRequest.newBuilder() + .setInput( + Payloads.newBuilder() + .addPayloads( + Payload.newBuilder().setData(ByteString.copyFrom(new byte[dataLen])).build())) + .build(); + } + + private static PayloadLimitsOptions warnAt(long size) { + return PayloadLimitsOptions.newBuilder() + .setPayloadsWarnSize(size) + .setMemoWarnSize(size) + .build(); + } + + @Test + public void oversizedRequestWithoutErrorLimitsIsSent() { + WorkflowServiceStubs stubs = stubs(warnAt(10)); + stubs.blockingStub().startWorkflowExecution(startWithInput(1000)); + assertEquals(1, startCount.get()); + } + + @Test + public void anOversizedRequestWithoutErrorLimitsIsLoggedAsAWarning() { + LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory(); + Logger validatorLog = loggerContext.getLogger(PAYLOAD_LIMIT_VALIDATOR_LOGGER); + ListAppender logs = new ListAppender<>(); + logs.setContext(loggerContext); + logs.start(); + validatorLog.addAppender(logs); + try { + stubs(warnAt(10)).blockingStub().startWorkflowExecution(startWithInput(1000)); + } finally { + validatorLog.detachAppender(logs); + logs.stop(); + } + + assertEquals(1, startCount.get()); + assertEquals(1, logs.list.size()); + ILoggingEvent event = logs.list.get(0); + assertEquals(Level.WARN, event.getLevel()); + assertTrue( + event.getFormattedMessage(), + event + .getFormattedMessage() + .startsWith( + "[TMPRL1103] Attempted to upload payloads with size that exceeded the warning" + + " limit.")); + } + + @Test + public void errorLimitsRejectOversizedRequestWithoutSendingIt() { + WorkflowServiceStubs stubs = stubs(warnAt(10)); + StatusRuntimeException e = + assertThrows( + StatusRuntimeException.class, + () -> + stubs + .blockingStub() + .withOption(PayloadErrorLimits.CALL_OPTIONS_KEY, ERROR_LIMITS) + .startWorkflowExecution(startWithInput(1000))); + + assertEquals(Status.Code.INVALID_ARGUMENT, e.getStatus().getCode()); + assertEquals( + "[TMPRL1103] Attempted to upload payloads with size that exceeded the error limit.", + e.getStatus().getDescription()); + PayloadLimitViolation violation = PayloadLimitViolationException.find(e).get(); + assertEquals("input", violation.getPath()); + assertEquals(100, violation.getLimit()); + assertEquals(0, startCount.get()); + } + + @Test + public void futureStubRejectsOversizedRequest() { + WorkflowServiceStubs stubs = stubs(PayloadLimitsOptions.getDefaultInstance()); + ExecutionException e = + assertThrows( + ExecutionException.class, + () -> + stubs + .futureStub() + .withOption(PayloadErrorLimits.CALL_OPTIONS_KEY, ERROR_LIMITS) + .startWorkflowExecution(startWithInput(1000)) + .get()); + assertTrue(PayloadLimitViolationException.find(e).isPresent()); + assertEquals(0, startCount.get()); + } + + @Test + public void errorLimitsAllowRequestUnderLimit() { + WorkflowServiceStubs stubs = stubs(PayloadLimitsOptions.getDefaultInstance()); + stubs + .blockingStub() + .withOption(PayloadErrorLimits.CALL_OPTIONS_KEY, ERROR_LIMITS) + .startWorkflowExecution(startWithInput(10)); + assertEquals(1, startCount.get()); + } + + @Test + public void noErrorLimitsSendAnOversizedRequest() { + WorkflowServiceStubs stubs = stubs(PayloadLimitsOptions.getDefaultInstance()); + stubs + .blockingStub() + .withOption(PayloadErrorLimits.CALL_OPTIONS_KEY, PayloadErrorLimits.NONE) + .startWorkflowExecution(startWithInput(1000)); + assertEquals(1, startCount.get()); + } + + @Test + public void requestWithoutPayloadFieldsIsUnaffected() { + WorkflowServiceStubs stubs = stubs(warnAt(1)); + stubs + .blockingStub() + .withOption(PayloadErrorLimits.CALL_OPTIONS_KEY, new PayloadErrorLimits(1, 1)) + .describeNamespace(DescribeNamespaceRequest.newBuilder().setNamespace("ns").build()); + assertEquals(1, describeCount.get()); + } + + @Test + public void requestChangedByUserInterceptorIsChecked() { + ClientInterceptor enlargingInterceptor = + new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new ForwardingClientCall.SimpleForwardingClientCall( + next.newCall(method, callOptions)) { + @Override + @SuppressWarnings("unchecked") + public void sendMessage(ReqT message) { + if (message instanceof StartWorkflowExecutionRequest) { + message = (ReqT) startWithInput(1000); + } + super.sendMessage(message); + } + }; + } + }; + WorkflowServiceStubs stubs = + stubs(PayloadLimitsOptions.getDefaultInstance(), enlargingInterceptor); + + assertThrows( + StatusRuntimeException.class, + () -> + stubs + .blockingStub() + .withOption(PayloadErrorLimits.CALL_OPTIONS_KEY, ERROR_LIMITS) + .startWorkflowExecution(startWithInput(10))); + assertEquals(0, startCount.get()); + } + + @Test + public void aRejectedRequestIsNotCountedAsARequest() throws Exception { + CountingReporter reporter = new CountingReporter(); + Scope scope = new RootScopeBuilder().reporter(reporter).reportEvery(Duration.ofSeconds(1)); + WorkflowServiceStubs stubs = + WorkflowServiceStubs.newServiceStubs( + WorkflowServiceStubsOptions.newBuilder() + .setChannel(channel) + .setMetricsScope(scope) + .build()); + + assertThrows( + StatusRuntimeException.class, + () -> + stubs + .blockingStub() + .withOption(PayloadErrorLimits.CALL_OPTIONS_KEY, ERROR_LIMITS) + .startWorkflowExecution(startWithInput(1000))); + stubs + .blockingStub() + .withOption(PayloadErrorLimits.CALL_OPTIONS_KEY, ERROR_LIMITS) + .startWorkflowExecution(startWithInput(10)); + scope.close(); + + assertEquals( + "only the request that was sent is counted", + 1, + reporter.count(MetricsType.TEMPORAL_REQUEST, "StartWorkflowExecution")); + } + + @Test + public void aRejectedOrEarlyCancelledCallNeverReachesTheNextInterceptor() { + Channel next = mock(Channel.class); + PayloadLimitsInterceptor interceptor = new PayloadLimitsInterceptor(10, 10); + CallOptions callOptions = + CallOptions.DEFAULT.withOption(PayloadErrorLimits.CALL_OPTIONS_KEY, ERROR_LIMITS); + + ClientCall rejected = + interceptor.interceptCall( + WorkflowServiceGrpc.getStartWorkflowExecutionMethod(), callOptions, next); + @SuppressWarnings("unchecked") + ClientCall.Listener rejectedListener = + mock(ClientCall.Listener.class); + rejected.start(rejectedListener, new Metadata()); + rejected.sendMessage(startWithInput(1000)); + rejected.halfClose(); + verify(rejectedListener) + .onClose(argThat(status -> status.getCode() == Status.Code.INVALID_ARGUMENT), any()); + + ClientCall cancelled = + interceptor.interceptCall( + WorkflowServiceGrpc.getStartWorkflowExecutionMethod(), callOptions, next); + @SuppressWarnings("unchecked") + ClientCall.Listener cancelledListener = + mock(ClientCall.Listener.class); + cancelled.start(cancelledListener, new Metadata()); + cancelled.cancel("deadline", null); + verify(cancelledListener) + .onClose(argThat(status -> status.getCode() == Status.Code.CANCELLED), any()); + + verify(next, never()).newCall(any(), any()); + } + + /** Records counter totals by metric name and operation. */ + @SuppressWarnings("deprecation") // StatsReporter's histogram methods take the deprecated Buckets. + private static final class CountingReporter implements StatsReporter { + private final Map counters = new ConcurrentHashMap<>(); + + long count(String name, String operation) { + return counters.getOrDefault(name + "|" + operation, 0L); + } + + @Override + public void reportCounter(String name, Map tags, long value) { + counters.merge(name + "|" + tags.get(MetricsTag.OPERATION_NAME), value, Long::sum); + } + + @Override + public void reportGauge(String name, Map tags, double value) {} + + @Override + public void reportTimer( + String name, Map tags, com.uber.m3.util.Duration interval) {} + + @Override + public void reportHistogramValueSamples( + String name, + Map tags, + com.uber.m3.tally.Buckets buckets, + double bucketLowerBound, + double bucketUpperBound, + long samples) {} + + @Override + public void reportHistogramDurationSamples( + String name, + Map tags, + com.uber.m3.tally.Buckets buckets, + com.uber.m3.util.Duration bucketLowerBound, + com.uber.m3.util.Duration bucketUpperBound, + long samples) {} + + @Override + public Capabilities capabilities() { + return CapableOf.REPORTING; + } + + @Override + public void flush() {} + + @Override + public void close() {} + } + + @Test + public void payloadLimitsOptionsDefaults() { + PayloadLimitsOptions defaults = PayloadLimitsOptions.getDefaultInstance(); + assertEquals(512 * 1024, defaults.getPayloadsWarnSize()); + assertEquals(2 * 1024, defaults.getMemoWarnSize()); + assertEquals(defaults, WorkflowServiceStubsOptions.getDefaultInstance().getPayloadLimits()); + } + + @Test + public void payloadLimitsOptionsRejectNegativeSizes() { + assertThrows( + IllegalArgumentException.class, + () -> PayloadLimitsOptions.newBuilder().setPayloadsWarnSize(-1).build()); + assertThrows( + IllegalArgumentException.class, + () -> PayloadLimitsOptions.newBuilder().setMemoWarnSize(-1).build()); + } + + @Test + public void workflowServiceStubsOptionsCopyPayloadLimits() { + WorkflowServiceStubsOptions options = + WorkflowServiceStubsOptions.newBuilder().setPayloadLimits(warnAt(7)).build(); + WorkflowServiceStubsOptions copy = WorkflowServiceStubsOptions.newBuilder(options).build(); + assertEquals(warnAt(7), copy.getPayloadLimits()); + assertEquals(options, copy); + assertNotEquals(options, WorkflowServiceStubsOptions.newBuilder().build()); + } +} diff --git a/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java b/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java index 8217007744..1d3b457d74 100644 --- a/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java +++ b/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java @@ -32,6 +32,7 @@ import io.temporal.internal.activity.ActivityTaskHandlerImpl; import io.temporal.internal.client.WorkflowClientInternal; import io.temporal.internal.common.ProtobufTimeUtils; +import io.temporal.internal.payload.limits.PayloadErrorLimits; import io.temporal.internal.payload.storage.ExternalStorageDataConverter; import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.internal.sync.*; @@ -124,7 +125,8 @@ public TestActivityEnvironmentInternal(@Nullable TestEnvironmentOptions options) WorkerOptions.getDefaultInstance().getDefaultHeartbeatThrottleInterval(), clientDataConverter, heartbeatExecutor, - externalStorageRunner); + externalStorageRunner, + () -> PayloadErrorLimits.NONE); activityTaskHandler = new ActivityTaskHandlerImpl( testEnvironmentOptions.getWorkflowClientOptions().getNamespace(),