diff --git a/CHANGELOG.md b/CHANGELOG.md index 4019816163..8a30365d81 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -51,6 +51,13 @@ to docs, or any other relevant information. `TEMPORAL_WORKFLOW_TASK_DURATION_WARN_SECONDS` environment variable. - Nexus operation handlers now forward inbound callback links to the workflows and signals they start, instead of dropping them. +- Nexus handlers now propagate `NexusSerializationContext` to the workflows and standalone + activities they start. These use it only for their results and failures, which are returned to + the Nexus caller. Standalone activities also use it for their heartbeat details, which the server + includes in their timeout failures. + To complete such an activity asynchronously, read the context from + `ActivityInfo.getNexusSerializationContext()`, store it with the task token, and pass it to + `ActivityCompletionClient.withContext(NexusSerializationContext)`. ### Changed - Moved the Spring AI integration to [temporalio/ai-integrations](https://github.com/temporalio/ai-integrations/tree/main/java/spring-ai). diff --git a/temporal-sdk/src/main/java/io/temporal/activity/ActivityInfo.java b/temporal-sdk/src/main/java/io/temporal/activity/ActivityInfo.java index ce42858748..c7f4b192dc 100644 --- a/temporal-sdk/src/main/java/io/temporal/activity/ActivityInfo.java +++ b/temporal-sdk/src/main/java/io/temporal/activity/ActivityInfo.java @@ -1,8 +1,10 @@ package io.temporal.activity; import io.temporal.api.common.v1.Payloads; +import io.temporal.common.Experimental; import io.temporal.common.Priority; import io.temporal.common.RetryOptions; +import io.temporal.payload.context.NexusSerializationContext; import java.time.Duration; import java.util.Optional; import javax.annotation.Nonnull; @@ -171,4 +173,19 @@ default boolean isInWorkflow() { * @return Retry options for the Activity Execution. */ RetryOptions getRetryOptions(); + + /** + * Returns the context of the Nexus operation handler that started this standalone activity, or + * {@code null} if it wasn't started by one. + * + *

To complete the activity asynchronously, store this context together with the {@link + * #getTaskToken() task token} and pass it to {@link + * io.temporal.client.ActivityCompletionClient#withContext(NexusSerializationContext)}, so the + * result is encoded with the context the Nexus caller decodes it with. + */ + @Experimental + @Nullable + default NexusSerializationContext getNexusSerializationContext() { + return null; + } } diff --git a/temporal-sdk/src/main/java/io/temporal/client/ActivityCompletionClient.java b/temporal-sdk/src/main/java/io/temporal/client/ActivityCompletionClient.java index 6e49690c20..9e7687722f 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/ActivityCompletionClient.java +++ b/temporal-sdk/src/main/java/io/temporal/client/ActivityCompletionClient.java @@ -1,11 +1,13 @@ package io.temporal.client; import io.temporal.activity.ActivityExecutionContext; +import io.temporal.activity.ActivityInfo; import io.temporal.common.Experimental; import io.temporal.common.converter.DataConverter; import io.temporal.common.converter.PayloadConverter; import io.temporal.payload.codec.PayloadCodec; import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.context.NexusSerializationContext; import io.temporal.payload.context.SerializationContext; import java.util.Optional; import javax.annotation.Nonnull; @@ -188,4 +190,19 @@ void heartbeatStandalone(String activityId, Optional activityRunId, @Experimental @Nonnull ActivityCompletionClient withContext(@Nonnull ActivitySerializationContext context); + + /** + * Supply this context when completing a standalone activity started by a Nexus operation handler, + * so its result, failure and heartbeat details are encoded with the context the Nexus caller + * decodes them with. Get it from {@link ActivityInfo#getNexusSerializationContext()} and store it + * together with the task token. + * + * @param context the Nexus context of the operation handler that started the activity + * @return an instance of ActivityCompletionClient that uses the provided {@code context} for + * serialization + * @see NexusSerializationContext + */ + @Experimental + @Nonnull + ActivityCompletionClient withContext(@Nonnull NexusSerializationContext context); } diff --git a/temporal-sdk/src/main/java/io/temporal/client/ActivityCompletionClientImpl.java b/temporal-sdk/src/main/java/io/temporal/client/ActivityCompletionClientImpl.java index 39fbb90829..6b139238da 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/ActivityCompletionClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/client/ActivityCompletionClientImpl.java @@ -4,6 +4,8 @@ import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.internal.client.external.ManualActivityCompletionClientFactory; import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.context.NexusSerializationContext; +import io.temporal.payload.context.SerializationContext; import io.temporal.workflow.Functions; import java.util.Optional; import javax.annotation.Nonnull; @@ -14,13 +16,13 @@ class ActivityCompletionClientImpl implements ActivityCompletionClient { private final ManualActivityCompletionClientFactory factory; private final Functions.Proc completionHandle; private final Scope metricsScope; - private final @Nullable ActivitySerializationContext serializationContext; + private final @Nullable SerializationContext serializationContext; ActivityCompletionClientImpl( ManualActivityCompletionClientFactory manualActivityCompletionClientFactory, Functions.Proc completionHandle, Scope metricsScope, - @Nullable ActivitySerializationContext serializationContext) { + @Nullable SerializationContext serializationContext) { this.factory = manualActivityCompletionClientFactory; this.completionHandle = completionHandle; this.metricsScope = metricsScope; @@ -146,6 +148,12 @@ public ActivityCompletionClient withContext(@Nonnull ActivitySerializationContex return new ActivityCompletionClientImpl(factory, completionHandle, metricsScope, context); } + @Nonnull + @Override + public ActivityCompletionClient withContext(@Nonnull NexusSerializationContext context) { + return new ActivityCompletionClientImpl(factory, completionHandle, metricsScope, context); + } + @Override public void heartbeatStandalone(String activityId, Optional activityRunId, V details) throws ActivityCompletionException { diff --git a/temporal-sdk/src/main/java/io/temporal/client/ActivityExecutionDescription.java b/temporal-sdk/src/main/java/io/temporal/client/ActivityExecutionDescription.java index ee67f0c0c4..c1a3cba0dd 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/ActivityExecutionDescription.java +++ b/temporal-sdk/src/main/java/io/temporal/client/ActivityExecutionDescription.java @@ -13,6 +13,7 @@ import io.temporal.internal.common.ProtobufTimeUtils; import io.temporal.internal.common.RetryOptionsUtils; import io.temporal.internal.common.SearchAttributesUtil; +import io.temporal.internal.nexus.PropagatedNexusSerializationContext; import io.temporal.payload.context.ActivitySerializationContext; import java.lang.reflect.Type; import java.time.Duration; @@ -29,6 +30,7 @@ public final class ActivityExecutionDescription extends ActivityExecutionMetadat private final DescribeActivityExecutionResponse response; private final DataConverter dataConverter; + private final DataConverter resultDataConverter; public ActivityExecutionDescription( DescribeActivityExecutionResponse response, DataConverter dataConverter, String namespace) { @@ -54,6 +56,16 @@ public ActivityExecutionDescription( dataConverter.withContext( new ActivitySerializationContext( namespace, null, null, getActivityType(), getTaskQueue(), false)); + // An activity started from a Nexus operation handler returns its result and failure to the + // Nexus caller, so they are serialized with the handler's Nexus context, which the server + // returns with the activity info. Heartbeat details use it too, because the server puts them + // into the timeout failure. Input and user metadata keep the activity context. + this.resultDataConverter = + response.getInfo().hasPropagatedNexusSerializationContext() + ? dataConverter.withContext( + PropagatedNexusSerializationContext.fromProto( + response.getInfo().getPropagatedNexusSerializationContext())) + : this.dataConverter; } private static @Nullable String nullIfEmpty(String s) { @@ -177,7 +189,7 @@ public boolean hasLastFailure() { @Nullable public RuntimeException getLastFailure() { return response.getInfo().hasLastFailure() - ? dataConverter.failureToException(response.getInfo().getLastFailure()) + ? resultDataConverter.failureToException(response.getInfo().getLastFailure()) : null; } @@ -252,7 +264,8 @@ public boolean hasHeartbeatDetails() { * requested without {@link DescribeActivityOptions.Builder#setIncludeHeartbeatDetails(boolean)}. */ public EncodedValues getHeartbeatDetails() { - return new EncodedValues(Optional.of(response.getInfo().getHeartbeatDetails()), dataConverter); + return new EncodedValues( + Optional.of(response.getInfo().getHeartbeatDetails()), resultDataConverter); } /** @@ -303,7 +316,7 @@ public Optional getResult(Class valueType, @Nullable Type genericType) return Optional.empty(); } return Optional.ofNullable( - dataConverter.fromPayloads( + resultDataConverter.fromPayloads( 0, Optional.of(response.getOutcome().getResult()), valueType, @@ -322,7 +335,7 @@ public RuntimeException getOutcomeFailure() { if (!response.getOutcome().hasFailure()) { return null; } - return dataConverter.failureToException(response.getOutcome().getFailure()); + return resultDataConverter.failureToException(response.getOutcome().getFailure()); } /** 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..38bc8c5525 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 @@ -11,7 +11,6 @@ import io.temporal.common.converter.DataConverter; import io.temporal.internal.client.external.ManualActivityCompletionClientFactory; import io.temporal.internal.payload.storage.ExternalStorageRunner; -import io.temporal.payload.context.ActivitySerializationContext; import io.temporal.workflow.Functions; import java.lang.reflect.Type; import java.time.Duration; @@ -155,13 +154,11 @@ public ManualActivityCompletionClient useLocalManualCompletion() { try { doNotCompleteOnReturn(); useLocalManualCompletion = true; - ActivitySerializationContext activitySerializationContext = - new ActivitySerializationContext(info); return new CompletionAwareManualCompletionClient( manualCompletionClientFactory.getClient( info.getTaskToken(), metricsScope, - activitySerializationContext, + HeartbeatContextImpl.heartbeatContext(info.getNamespace(), info), HeartbeatContextImpl.storageTargetForActivity(info.getNamespace(), info)), completionHandle); } finally { diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityInfoImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityInfoImpl.java index 2d7b83c274..2fbd215985 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityInfoImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityInfoImpl.java @@ -9,12 +9,15 @@ import io.temporal.internal.common.ProtoConverters; import io.temporal.internal.common.ProtobufTimeUtils; import io.temporal.internal.common.RetryOptionsUtils; +import io.temporal.internal.nexus.PropagatedNexusSerializationContext; +import io.temporal.payload.context.NexusSerializationContext; import io.temporal.workflow.Functions; import java.time.Duration; import java.util.Base64; import java.util.Objects; import java.util.Optional; import javax.annotation.Nonnull; +import javax.annotation.Nullable; final class ActivityInfoImpl implements ActivityInfoInternal { private final String namespace; @@ -205,6 +208,15 @@ public Optional

getHeader() { return Optional.empty(); } + @Nullable + @Override + public NexusSerializationContext getNexusSerializationContext() { + return response.hasPropagatedNexusSerializationContext() + ? PropagatedNexusSerializationContext.fromProto( + response.getPropagatedNexusSerializationContext()) + : null; + } + @Override public String toString() { return "ActivityInfo{" 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..50ade355f9 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 @@ -19,6 +19,7 @@ import io.temporal.internal.common.FailureUtils; import io.temporal.internal.worker.ActivityTaskHandler; import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.context.NexusSerializationContext; import io.temporal.serviceclient.CheckedExceptionWrapper; import java.lang.reflect.Method; import java.util.HashMap; @@ -69,6 +70,14 @@ public ActivityTaskHandler.Result execute(ActivityInfoInternal info, Scope metri activityInfo.isLocal()); DataConverter dataConverterWithActivityContext = dataConverter.withContext(serializationContext); + // The result and failure of a standalone activity started from a Nexus operation handler are + // returned to the Nexus caller, so they use the handler's Nexus context. Heartbeat details + // use it too, because the server puts them into the timeout failure. + NexusSerializationContext nexusContext = info.getNexusSerializationContext(); + DataConverter resultDataConverter = + nexusContext != null + ? dataConverter.withContext(nexusContext) + : dataConverterWithActivityContext; try { info.getHeader() @@ -92,7 +101,7 @@ public ActivityTaskHandler.Result execute(ActivityInfoInternal info, Scope metri info.getActivityId(), null, null, null, context.isUseLocalManualCompletion()); } - return this.constructSuccessfulResultValue(info, result, dataConverterWithActivityContext); + return this.constructSuccessfulResultValue(info, result, resultDataConverter); } catch (Throwable e) { Throwable ex = CheckedExceptionWrapper.unwrap(e); boolean local = info.isLocal(); @@ -127,7 +136,7 @@ public ActivityTaskHandler.Result execute(ActivityInfoInternal info, Scope metri context.getLastHeartbeatValue(), metricsScope, local, - dataConverterWithActivityContext); + resultDataConverter); } finally { if (context.isDoNotCompleteOnReturn()) { if (!context.isUseLocalManualCompletion()) { @@ -150,19 +159,16 @@ abstract Object[] provideArgs( Optional input, DataConverter dataConverterWithActivityContext); protected abstract ActivityTaskHandler.Result constructSuccessfulResultValue( - ActivityInfoInternal info, - ActivityOutput result, - DataConverter dataConverterWithActivityContext); + ActivityInfoInternal info, ActivityOutput result, DataConverter resultDataConverter); ActivityTaskHandler.Result constructResultValue( ActivityInfoInternal info, @Nullable ActivityOutput result, - DataConverter dataConverterWithActivityContext) { + DataConverter resultDataConverter) { RespondActivityTaskCompletedRequest.Builder request = RespondActivityTaskCompletedRequest.newBuilder(); if (result != null) { - Optional serialized = - dataConverterWithActivityContext.toPayloads(result.getResult()); + Optional serialized = resultDataConverter.toPayloads(result.getResult()); serialized.ifPresent(request::setResult); } return new ActivityTaskHandler.Result( @@ -218,14 +224,12 @@ Object[] provideArgs(Optional input, DataConverter dataConverterWithAc @Override protected ActivityTaskHandler.Result constructSuccessfulResultValue( - ActivityInfoInternal info, - ActivityOutput result, - DataConverter dataConverterWithActivityContext) { + ActivityInfoInternal info, ActivityOutput result, DataConverter resultDataConverter) { return constructResultValue( info, // if the expected result of the method is null, we don't publish result at all method.getReturnType() != Void.TYPE ? result : null, - dataConverterWithActivityContext); + resultDataConverter); } } @@ -261,10 +265,8 @@ Object[] provideArgs(Optional input, DataConverter dataConverterWithAc @Override protected ActivityTaskHandler.Result constructSuccessfulResultValue( - ActivityInfoInternal info, - ActivityOutput result, - DataConverter dataConverterWithActivityContext) { - return constructResultValue(info, result, dataConverterWithActivityContext); + ActivityInfoInternal info, ActivityOutput result, DataConverter resultDataConverter) { + return constructResultValue(info, result, resultDataConverter); } } } 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..666f24d850 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 @@ -21,6 +21,8 @@ import io.temporal.internal.payload.storage.ActivityStorageTargets; import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.context.NexusSerializationContext; +import io.temporal.payload.context.SerializationContext; import io.temporal.payload.storage.StorageDriverTargetInfo; import io.temporal.serviceclient.WorkflowServiceStubs; import java.lang.reflect.Type; @@ -77,7 +79,7 @@ static long getLocalHeartbeatTimeoutBufferMillis() { private final ScheduledExecutorService heartbeatExecutor; private final long heartbeatIntervalMillis; private final DataConverter dataConverter; - private final DataConverter dataConverterWithActivityContext; + private final DataConverter heartbeatDataConverter; private final @Nullable ExternalStorageRunner externalStorage; private final Scope metricsScope; @@ -145,15 +147,7 @@ public HeartbeatContextImpl( this.metricsScope = metricsScope; this.dataConverter = dataConverter; this.externalStorage = externalStorage; - this.dataConverterWithActivityContext = - dataConverter.withContext( - new ActivitySerializationContext( - namespace, - info.getWorkflowId(), - info.getWorkflowType(), - info.getActivityType(), - info.getActivityTaskQueue(), - info.isLocal())); + this.heartbeatDataConverter = dataConverter.withContext(heartbeatContext(namespace, info)); this.namespace = namespace; this.info = info; this.identity = identity; @@ -215,7 +209,7 @@ public Optional getHeartbeatDetails(Class detailsClass, Type detailsGe return Optional.ofNullable((V) this.lastDetails); } else { return Optional.ofNullable( - dataConverterWithActivityContext.fromPayloads( + heartbeatDataConverter.fromPayloads( 0, prevAttemptHeartbeatDetails, detailsClass, detailsGenericType)); } } finally { @@ -232,7 +226,7 @@ public Optional getLastHeartbeatDetails(Class detailsClass, Type detai lock.lock(); try { return Optional.ofNullable( - dataConverterWithActivityContext.fromPayloads( + heartbeatDataConverter.fromPayloads( 0, prevAttemptHeartbeatDetails, detailsClass, detailsGenericType)); } finally { lock.unlock(); @@ -371,6 +365,24 @@ private StorageDriverTargetInfo activityStorageTarget() { return storageTargetForActivity(namespace, info); } + /** + * Returns the handler's Nexus context for a standalone activity started from a Nexus operation + * handler, and the activity context otherwise. + */ + static SerializationContext heartbeatContext(String namespace, ActivityInfo info) { + NexusSerializationContext nexusContext = info.getNexusSerializationContext(); + if (nexusContext != null) { + return nexusContext; + } + return new ActivitySerializationContext( + namespace, + info.getWorkflowId(), + info.getWorkflowType(), + info.getActivityType(), + info.getActivityTaskQueue(), + info.isLocal()); + } + /** * Standalone activities target the activity; workflow activities target their workflow, matching * where {@link io.temporal.internal.worker.ActivityWorker} stores the activity task payloads. A @@ -451,7 +463,7 @@ private void sendHeartbeatRequest(Object details) { .setTaskToken(ByteString.copyFrom(info.getTaskToken())) .setNamespace(namespace) .setIdentity(identity); - dataConverterWithActivityContext.toPayloads(details).ifPresent(builder::setDetails); + heartbeatDataConverter.toPayloads(details).ifPresent(builder::setDetails); if (externalStorage != null) { offloadHeartbeat(builder); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/ClientDataConverterFactory.java b/temporal-sdk/src/main/java/io/temporal/internal/client/ClientDataConverterFactory.java index b357a88531..0a471c5369 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/ClientDataConverterFactory.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/ClientDataConverterFactory.java @@ -5,6 +5,7 @@ import io.temporal.internal.common.converter.TemporalTransferTypeDataConverter; import io.temporal.internal.payload.storage.ExternalStorageDataConverter; import io.temporal.internal.payload.storage.ExternalStorageRunner; +import io.temporal.payload.context.SerializationContext; import io.temporal.payload.context.WorkflowSerializationContext; import io.temporal.payload.storage.StorageDriverWorkflowInfo; import java.util.Objects; @@ -47,8 +48,21 @@ private static DataConverter transferAware(DataConverter configured) { DataConverter forWorkflow( String workflowId, @Nullable String runId, @Nullable String workflowType) { - DataConverter converter = - workflowConverter.withContext(new WorkflowSerializationContext(namespace, workflowId)); + return forWorkflow( + new WorkflowSerializationContext(namespace, workflowId), workflowId, runId, workflowType); + } + + /** + * Same as {@link #forWorkflow(String, String, String)}, but with an explicit serialization + * context. Used for payloads of a workflow started by a Nexus operation handler, which are + * serialized with the handler's Nexus context rather than the workflow's own. + */ + DataConverter forWorkflow( + SerializationContext context, + String workflowId, + @Nullable String runId, + @Nullable String workflowType) { + DataConverter converter = workflowConverter.withContext(context); return ((ExternalStorageDataConverter) converter) .withStorageTarget( new StorageDriverWorkflowInfo( diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java b/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java index 50a65dc07c..6c8f801cf6 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java @@ -16,6 +16,7 @@ import io.temporal.api.common.v1.Link; import io.temporal.api.common.v1.Payloads; import io.temporal.api.errordetails.v1.ActivityExecutionAlreadyStartedFailure; +import io.temporal.api.nexus.v1.PropagatedSerializationContext; import io.temporal.api.sdk.v1.UserMetadata; import io.temporal.api.taskqueue.v1.TaskQueue; import io.temporal.api.workflowservice.v1.*; @@ -32,7 +33,9 @@ import io.temporal.internal.nexus.CurrentNexusOperationContext; import io.temporal.internal.nexus.InternalNexusOperationContext; import io.temporal.internal.nexus.NexusOperationMetadata; +import io.temporal.internal.nexus.PropagatedNexusSerializationContext; import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.context.SerializationContext; import io.temporal.serviceclient.StatusUtils; import java.lang.reflect.Type; import java.util.*; @@ -40,6 +43,7 @@ import java.util.concurrent.CompletionException; import java.util.concurrent.TimeoutException; import java.util.stream.StreamSupport; +import javax.annotation.Nullable; /** * Terminus of the activity interceptor chain. Implements all activity RPCs against the Temporal @@ -69,6 +73,13 @@ public StartActivityOutput startActivity(StartActivityInput input) { if (Strings.isNullOrEmpty(options.getTaskQueue())) { throw new IllegalArgumentException("taskQueue must not be null or empty"); } + // An activity started from a Nexus operation handler returns its result and failure to the + // Nexus caller, so they are serialized with the handler's Nexus context. It is sent with the + // request so the server can propagate it to the activity worker and to clients reading the + // result. The input and user metadata keep the activity context. + @Nullable + PropagatedSerializationContext propagatedNexusSerializationContext = + PropagatedNexusSerializationContext.current(); DataConverter dc = dataConverter.withContext( new ActivitySerializationContext( @@ -94,6 +105,9 @@ public StartActivityOutput startActivity(StartActivityInput input) { .setTaskQueue(TaskQueue.newBuilder().setName(options.getTaskQueue()).build()) .setIdReusePolicy(options.getIdReusePolicy()) .setIdConflictPolicy(options.getIdConflictPolicy()); + if (propagatedNexusSerializationContext != null) { + request.setPropagatedNexusSerializationContext(propagatedNexusSerializationContext); + } Optional activityInput = dc.toPayloads(input.getArgs().toArray()); activityInput.ifPresent(request::setInput); @@ -212,7 +226,6 @@ public StartActivityOutput startActivity(StartActivityInput input) { @Override public GetActivityResultOutput getActivityResult(GetActivityResultInput input) throws TimeoutException { - DataConverter dc = dataConverter.withContext(resultSerializationContext(input)); Deadline deadline = Deadline.after(input.getTimeout(), input.getTimeoutUnit()); while (true) { @@ -250,6 +263,7 @@ public GetActivityResultOutput getActivityResult(GetActivityResultInput GetActivityResultOutput getActivityResult(GetActivityResultInput CompletableFuture> getActivityResultAsync( GetActivityResultInput input) { - DataConverter dc = dataConverter.withContext(resultSerializationContext(input)); Deadline deadline = Deadline.after(input.getTimeout(), input.getTimeoutUnit()); return pollActivityUntilOutcome(input, deadline) .handle( - (outcome, e) -> { + (response, e) -> { if (e == null) { - return decodeOutcome(outcome, input, dc); + return decodeOutcome(response.getOutcome(), input, resultDataConverter(response)); } throw handleAsyncException(e, deadline, input.getActivityId()); }); } - private CompletableFuture pollActivityUntilOutcome( + private CompletableFuture pollActivityUntilOutcome( GetActivityResultInput input, Deadline deadline) { PollActivityExecutionRequest.Builder pollRequest = PollActivityExecutionRequest.newBuilder() @@ -315,7 +328,7 @@ private CompletableFuture pollActivityUntilOutcome( if (!response.hasOutcome()) { return pollActivityUntilOutcome(input, deadline); } - return CompletableFuture.completedFuture(response.getOutcome()); + return CompletableFuture.completedFuture(response); }); } @@ -369,10 +382,20 @@ private GetActivityResultOutput decodeOutcome( } } - private ActivitySerializationContext resultSerializationContext(GetActivityResultInput input) { - // Currently, result doesn't have access to activity type and serialization context doesn't hold - // activity ID. - return new ActivitySerializationContext(namespace, null, null, null, null, false); + /** + * The data converter for the result or failure in a poll response. An activity started from a + * Nexus operation handler uses the handler's Nexus context, which the server returns with the + * response; other activities use an activity context. + */ + private DataConverter resultDataConverter(PollActivityExecutionResponse response) { + SerializationContext context = + response.hasPropagatedNexusSerializationContext() + ? PropagatedNexusSerializationContext.fromProto( + response.getPropagatedNexusSerializationContext()) + // Currently, result doesn't have access to activity type and serialization context + // doesn't hold activity ID. + : new ActivitySerializationContext(namespace, null, null, null, null, false); + return dataConverter.withContext(context); } @Override diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/RootNexusClientInvoker.java b/temporal-sdk/src/main/java/io/temporal/internal/client/RootNexusClientInvoker.java index d92e2f3989..6299305119 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/RootNexusClientInvoker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/RootNexusClientInvoker.java @@ -34,6 +34,7 @@ import io.temporal.internal.client.external.GenericWorkflowClient; import io.temporal.internal.common.ProtobufTimeUtils; import io.temporal.internal.common.WorkflowExecutionUtils; +import io.temporal.internal.nexus.PropagatedNexusSerializationContext; import io.temporal.payload.context.NexusSerializationContext; import io.temporal.serviceclient.StatusUtils; import java.util.Iterator; @@ -164,16 +165,26 @@ public DescribeNexusOperationExecutionOutput describeNexusOperationExecution( } /** - * The client's data converter scoped to the Nexus operation the result is being read for, or left - * as-is when the operation is unknown, which is the case for a handle obtained by operation ID. + * The client's data converter scoped to the Nexus operation the result is being read for. + * + *

Prefers the context the server returns with the poll response, which identifies the + * operation even for a handle obtained by operation ID. Older servers don't return it, so this + * falls back to the context of the handle's start request, or leaves the converter as-is when the + * operation is unknown, which is the case for a handle obtained by operation ID. * *

{@link GetNexusOperationResultInput} guarantees the endpoint, service and operation are set * together or not at all, so one null means all three are null. Absence is tested with {@code * null} rather than emptiness so an operation genuinely named with an empty string still gets a * context. */ - private DataConverter dataConverterFor(GetNexusOperationResultInput input) { + private DataConverter dataConverterFor( + GetNexusOperationResultInput input, PollNexusOperationExecutionResponse response) { DataConverter dataConverter = clientOptions.getDataConverter(); + if (response.hasPropagatedNexusSerializationContext()) { + return dataConverter.withContext( + PropagatedNexusSerializationContext.fromProto( + response.getPropagatedNexusSerializationContext())); + } if (input.getEndpoint() == null) { return dataConverter; } @@ -283,7 +294,7 @@ private GetNexusOperationResultOutput extractResult( @Nullable String runId, PollNexusOperationExecutionResponse response, GetNexusOperationResultInput input) { - DataConverter dataConverter = dataConverterFor(input); + DataConverter dataConverter = dataConverterFor(input, response); if (response.hasFailure()) { Failure failure = response.getFailure(); throw new NexusOperationFailedException( diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java b/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java index ad56de7812..e2284aad2e 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java @@ -17,6 +17,8 @@ import io.temporal.api.enums.v1.WorkflowExecutionStatus; import io.temporal.api.errordetails.v1.MultiOperationExecutionFailure; import io.temporal.api.failure.v1.MultiOperationExecutionAborted; +import io.temporal.api.history.v1.HistoryEvent; +import io.temporal.api.nexus.v1.PropagatedSerializationContext; import io.temporal.api.query.v1.WorkflowQuery; import io.temporal.api.sdk.v1.UserMetadata; import io.temporal.api.update.v1.*; @@ -29,10 +31,12 @@ import io.temporal.internal.client.external.GenericWorkflowClient; import io.temporal.internal.common.HeaderUtils; import io.temporal.internal.common.InternalUtils; +import io.temporal.internal.common.WorkflowExecutionUtils; import io.temporal.internal.nexus.CurrentNexusOperationContext; import io.temporal.internal.nexus.InternalNexusOperationContext; import io.temporal.internal.nexus.NexusOperationMetadata; import io.temporal.internal.nexus.OperationTokenUtil; +import io.temporal.internal.nexus.PropagatedNexusSerializationContext; import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.internal.worker.WorkerVersioningProtoUtils; import io.temporal.payload.storage.StorageDriverWorkflowInfo; @@ -139,7 +143,8 @@ public WorkflowStartOutput start(WorkflowStartInput input) { workflowConverter(input.getWorkflowId(), null, input.getWorkflowType()); StartWorkflowExecutionRequest.Builder startRequest = - toStartRequest(dataConverterWithWorkflowContext, input); + toStartRequest( + dataConverterWithWorkflowContext, input, PropagatedNexusSerializationContext.current()); try (@Nullable WorkflowTaskDispatchHandle eagerDispatchHandle = obtainDispatchHandle(input)) { boolean requestEagerExecution = eagerDispatchHandle != null; @@ -232,7 +237,7 @@ public WorkflowSignalWithStartOutput signalWithStart(WorkflowSignalWithStartInpu workflowConverter( workflowStartInput.getWorkflowId(), null, workflowStartInput.getWorkflowType()); StartWorkflowExecutionRequestOrBuilder startRequest = - toStartRequest(dataConverterWithWorkflowContext, workflowStartInput); + toStartRequest(dataConverterWithWorkflowContext, workflowStartInput, null); Optional signalInput = dataConverterWithWorkflowContext.toPayloads(input.getSignalArguments()); @@ -270,6 +275,9 @@ public WorkflowUpdateWithStartOutput updateWithStart( WorkflowStartInput startInput = input.getWorkflowStartInput(); DataConverter dataConverterWithWorkflowContext = workflowConverter(startInput.getWorkflowId(), null, startInput.getWorkflowType()); + @Nullable + PropagatedSerializationContext propagatedNexusSerializationContext = + PropagatedNexusSerializationContext.current(); ExecuteMultiOperationRequest request = ExecuteMultiOperationRequest.newBuilder() @@ -277,7 +285,11 @@ public WorkflowUpdateWithStartOutput updateWithStart( .addOperations( 0, ExecuteMultiOperationRequest.Operation.newBuilder() - .setStartWorkflow(toStartRequest(dataConverterWithWorkflowContext, startInput)) + .setStartWorkflow( + toStartRequest( + dataConverterWithWorkflowContext, + startInput, + propagatedNexusSerializationContext)) .build()) .addOperations( 1, @@ -373,7 +385,10 @@ public WorkflowUpdateWithStartOutput updateWithStart( WorkflowExecution.newBuilder() .setRunId(startResponse.getRunId()) .setWorkflowId( - toStartRequest(dataConverterWithWorkflowContext, startInput) + toStartRequest( + dataConverterWithWorkflowContext, + startInput, + propagatedNexusSerializationContext) .build() .getWorkflowId()) .build(); @@ -381,7 +396,9 @@ public WorkflowUpdateWithStartOutput updateWithStart( } private StartWorkflowExecutionRequest.Builder toStartRequest( - DataConverter dataConverterWithWorkflowContext, WorkflowStartInput workflowStartInput) { + DataConverter dataConverterWithWorkflowContext, + WorkflowStartInput workflowStartInput, + @Nullable PropagatedSerializationContext propagatedNexusSerializationContext) { Optional workflowInput = dataConverterWithWorkflowContext.toPayloads(workflowStartInput.getArguments()); @@ -412,6 +429,9 @@ private StartWorkflowExecutionRequest.Builder toStartRequest( workflowInput.orElse(null), memo, userMetadata); + if (propagatedNexusSerializationContext != null) { + startRequest.setPropagatedNexusSerializationContext(propagatedNexusSerializationContext); + } storeHeader( startRequest.getHeaderBuilder(), workflowStartInput.getWorkflowId(), @@ -422,46 +442,54 @@ private StartWorkflowExecutionRequest.Builder toStartRequest( @Override public GetResultOutput getResult(GetResultInput input) throws TimeoutException { - DataConverter dataConverterWithWorkflowContext = - workflowConverter(input.getWorkflowExecution()); - Optional resultValue = - WorkflowClientLongPollHelper.getWorkflowExecutionResult( + GetWorkflowExecutionHistoryResponse closed = + WorkflowClientLongPollHelper.getInstanceCloseEvent( genericClient, requestsHelper, input.getWorkflowExecution(), - input.getWorkflowType(), - dataConverterWithWorkflowContext, input.getTimeout(), input.getTimeoutUnit()); - return new GetResultOutput<>( - convertResultPayloads( - resultValue, - input.getResultClass(), - input.getResultType(), - dataConverterWithWorkflowContext)); + return new GetResultOutput<>(convertClosedWorkflowResult(input, closed)); } @Override public GetResultAsyncOutput getResultAsync(GetResultInput input) { - DataConverter dataConverterWithWorkflowContext = - workflowConverter(input.getWorkflowExecution()); - CompletableFuture> resultValue = - WorkflowClientLongPollAsyncHelper.getWorkflowExecutionResultAsync( + CompletableFuture closed = + WorkflowClientLongPollAsyncHelper.getInstanceCloseEventAsync( genericClient, requestsHelper, input.getWorkflowExecution(), - input.getWorkflowType(), input.getTimeout(), - input.getTimeoutUnit(), - dataConverterWithWorkflowContext); - return new GetResultAsyncOutput<>( - resultValue.thenApply( - payloads -> - convertResultPayloads( - payloads, - input.getResultClass(), - input.getResultType(), - dataConverterWithWorkflowContext))); + input.getTimeoutUnit()); + return new GetResultAsyncOutput<>(closed.thenApply(c -> convertClosedWorkflowResult(input, c))); + } + + /** + * Decodes the result of a closed workflow, or throws the failure it closed with. + * + *

A workflow started by a Nexus operation handler encodes its result and failure with the + * handler's Nexus serialization context, which the server returns with the history. Other + * workflows use their own workflow context. + */ + private R convertClosedWorkflowResult( + GetResultInput input, GetWorkflowExecutionHistoryResponse closed) { + WorkflowExecution execution = input.getWorkflowExecution(); + DataConverter dataConverter = + closed.hasPropagatedNexusSerializationContext() + ? converterFactory.forWorkflow( + PropagatedNexusSerializationContext.fromProto( + closed.getPropagatedNexusSerializationContext()), + execution.getWorkflowId(), + execution.getRunId(), + null) + : workflowConverter(execution); + // The long poll helpers return a response whose single event is the close event. + HistoryEvent closeEvent = closed.getHistory().getEvents(0); + Optional resultValue = + WorkflowExecutionUtils.getResultFromCloseEvent( + execution, input.getWorkflowType(), closeEvent, dataConverter); + return convertResultPayloads( + resultValue, input.getResultClass(), input.getResultType(), dataConverter); } @Override diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientLongPollAsyncHelper.java b/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientLongPollAsyncHelper.java index 15830aa7e3..58a53fd632 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientLongPollAsyncHelper.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientLongPollAsyncHelper.java @@ -1,48 +1,49 @@ package io.temporal.internal.client; -import static io.temporal.internal.common.WorkflowExecutionUtils.getResultFromCloseEvent; - import com.google.protobuf.ByteString; import io.grpc.Deadline; import io.grpc.Status; import io.grpc.StatusRuntimeException; -import io.temporal.api.common.v1.Payloads; import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.api.enums.v1.EventType; import io.temporal.api.history.v1.History; import io.temporal.api.history.v1.HistoryEvent; +import io.temporal.api.nexus.v1.PropagatedSerializationContext; import io.temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryRequest; import io.temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse; -import io.temporal.common.converter.DataConverter; import io.temporal.internal.client.external.GenericWorkflowClient; import io.temporal.internal.common.WorkflowExecutionUtils; -import java.util.Optional; import java.util.concurrent.*; import javax.annotation.Nonnull; +import javax.annotation.Nullable; /** This class encapsulates async long poll logic of {@link RootWorkflowClientInvoker} */ final class WorkflowClientLongPollAsyncHelper { - static CompletableFuture> getWorkflowExecutionResultAsync( + /** + * Returns the history response whose single event closed the workflow, potentially waiting for + * workflow or continue-as-new executions to complete. It carries the Nexus serialization context + * of the closed run, if the server returned one on any page of that run. Completes exceptionally + * with a {@link TimeoutException} if the workflow didn't complete within the specified timeout. + */ + static CompletableFuture getInstanceCloseEventAsync( GenericWorkflowClient genericClient, WorkflowClientRequestFactory workflowClientHelper, @Nonnull WorkflowExecution workflowExecution, - Optional workflowType, long timeout, - TimeUnit unit, - DataConverter converter) { + TimeUnit unit) { Deadline longPollTimeoutDeadline = Deadline.after(timeout, unit); return getInstanceCloseEventAsync( genericClient, workflowClientHelper, workflowExecution, ByteString.EMPTY, + null, longPollTimeoutDeadline) .handle( - (closeEvent, e) -> { + (closed, e) -> { if (e == null) { - return getResultFromCloseEvent( - workflowExecution, workflowType, closeEvent, converter); + return closed; } else { throw handleException(e, longPollTimeoutDeadline, workflowExecution, timeout, unit); } @@ -74,12 +75,18 @@ private static CompletionException handleException( } } - /** Returns an instance closing event, potentially waiting for workflow to complete. */ - private static CompletableFuture getInstanceCloseEventAsync( + /** + * Returns an instance closing event, potentially waiting for workflow to complete. + * + * @param propagated the Nexus serialization context returned so far for the current run, carried + * across empty long polls of the same run. + */ + private static CompletableFuture getInstanceCloseEventAsync( GenericWorkflowClient genericClient, WorkflowClientRequestFactory workflowClientHelper, final WorkflowExecution workflowExecution, ByteString pageToken, + @Nullable PropagatedSerializationContext propagated, Deadline longPollTimeoutDeadline) { GetWorkflowExecutionHistoryRequest request = workflowClientHelper.newHistoryLongPollRequest(workflowExecution, pageToken); @@ -87,6 +94,10 @@ private static CompletableFuture getInstanceCloseEventAsync( genericClient.longPollHistoryAsync(request, longPollTimeoutDeadline); return response.thenComposeAsync( (r) -> { + PropagatedSerializationContext runPropagated = + r.hasPropagatedNexusSerializationContext() + ? r.getPropagatedNexusSerializationContext() + : propagated; History history = r.getHistory(); if (history.getEventsCount() == 0) { // Empty poll returned @@ -97,6 +108,7 @@ private static CompletableFuture getInstanceCloseEventAsync( workflowClientHelper, workflowExecution, nextPageToken, + runPropagated, longPollTimeoutDeadline); } HistoryEvent event = history.getEvents(0); // should be only one event @@ -118,9 +130,12 @@ private static CompletableFuture getInstanceCloseEventAsync( workflowClientHelper, nextWorkflowExecution, ByteString.EMPTY, + null, longPollTimeoutDeadline); } - return CompletableFuture.completedFuture(event); + return CompletableFuture.completedFuture( + WorkflowClientLongPollHelper.withPropagatedNexusSerializationContext( + r, runPropagated)); }); } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientLongPollHelper.java b/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientLongPollHelper.java index 77f20fa6c9..961741806c 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientLongPollHelper.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientLongPollHelper.java @@ -4,61 +4,33 @@ import io.grpc.Deadline; import io.grpc.Status; import io.grpc.StatusRuntimeException; -import io.temporal.api.common.v1.Payloads; import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.api.enums.v1.*; import io.temporal.api.history.v1.*; +import io.temporal.api.nexus.v1.PropagatedSerializationContext; import io.temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryRequest; import io.temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse; -import io.temporal.client.WorkflowFailedException; -import io.temporal.common.converter.DataConverter; -import io.temporal.failure.CanceledFailure; import io.temporal.internal.client.external.GenericWorkflowClient; import io.temporal.internal.common.WorkflowExecutionUtils; -import java.util.Optional; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import javax.annotation.Nonnull; +import javax.annotation.Nullable; /** This class encapsulates sync long poll logic of {@link RootWorkflowClientInvoker} */ final class WorkflowClientLongPollHelper { private WorkflowClientLongPollHelper() {} - /** - * Returns result of a workflow instance execution or throws an exception if workflow did not - * complete successfully. Will wait for continue-as-new executions of the original workflow - * execution if present. - * - * @param workflowType is optional. - * @throws TimeoutException if workflow didn't complete within specified timeout - * @throws CanceledFailure if workflow was canceled - * @throws WorkflowFailedException if workflow execution failed - */ - static Optional getWorkflowExecutionResult( - GenericWorkflowClient genericClient, - WorkflowClientRequestFactory workflowClientHelper, - @Nonnull WorkflowExecution workflowExecution, - Optional workflowType, - DataConverter converter, - long timeout, - TimeUnit unit) - throws TimeoutException { - // getInstanceCloseEvent waits for workflow completion including new runs. - HistoryEvent closeEvent = - getInstanceCloseEvent( - genericClient, workflowClientHelper, workflowExecution, timeout, unit); - return WorkflowExecutionUtils.getResultFromCloseEvent( - workflowExecution, workflowType, closeEvent, converter); - } - /** * @param timeout timeout to retrieve InstanceCloseEvent in {@code unit} units. If 0 - MAX_INTEGER * will be used * @param unit time unit of {@code timeout} - * @return an instance closing event, potentially waiting for workflow or continue-as-new - * executions to complete. + * @return the history response whose single event closed the workflow, potentially waiting for + * workflow or continue-as-new executions to complete. It carries the Nexus serialization + * context of the closed run, if the server returned one on any page of that run. + * @throws TimeoutException if workflow didn't complete within specified timeout */ - private static HistoryEvent getInstanceCloseEvent( + static GetWorkflowExecutionHistoryResponse getInstanceCloseEvent( GenericWorkflowClient genericClient, WorkflowClientRequestFactory workflowClientHelper, @Nonnull WorkflowExecution workflowExecution, @@ -68,6 +40,8 @@ private static HistoryEvent getInstanceCloseEvent( ByteString pageToken = ByteString.EMPTY; GetWorkflowExecutionHistoryResponse response; Deadline longPollTimeoutDeadline = Deadline.after(timeout, unit); + // Each run of a continue-as-new chain carries its own context, so it is reset per run. + @Nullable PropagatedSerializationContext propagatedSerializationContext = null; while (true) { GetWorkflowExecutionHistoryRequest request = @@ -85,6 +59,9 @@ private static HistoryEvent getInstanceCloseEvent( } throw e; } + if (response.hasPropagatedNexusSerializationContext()) { + propagatedSerializationContext = response.getPropagatedNexusSerializationContext(); + } History history = response.getHistory(); if (history.getEventsCount() > 0) { @@ -95,6 +72,7 @@ private static HistoryEvent getInstanceCloseEvent( // Workflow called continueAsNew. Start polling the new execution with new runId. if (event.getEventType() == EventType.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW) { pageToken = ByteString.EMPTY; + propagatedSerializationContext = null; workflowExecution = WorkflowExecution.newBuilder() .setWorkflowId(workflowExecution.getWorkflowId()) @@ -105,7 +83,7 @@ private static HistoryEvent getInstanceCloseEvent( .build(); continue; } - return event; + return withPropagatedNexusSerializationContext(response, propagatedSerializationContext); } if (!response.getNextPageToken().isEmpty()) { pageToken = response.getNextPageToken(); @@ -113,6 +91,19 @@ private static HistoryEvent getInstanceCloseEvent( } } + /** + * Sets the Nexus serialization context retained from an earlier page of the run on the response + * that closed it, since paginated responses may omit it. + */ + static GetWorkflowExecutionHistoryResponse withPropagatedNexusSerializationContext( + GetWorkflowExecutionHistoryResponse response, + @Nullable PropagatedSerializationContext propagated) { + if (propagated == null || response.hasPropagatedNexusSerializationContext()) { + return response; + } + return response.toBuilder().setPropagatedNexusSerializationContext(propagated).build(); + } + static TimeoutException newTimeoutException( @Nonnull WorkflowExecution workflowExecution, long timeout, TimeUnit unit) { return new TimeoutException( diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactory.java b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactory.java index 327a5cb6d4..d86df2ec8e 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactory.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactory.java @@ -5,7 +5,7 @@ import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.common.converter.DataConverter; import io.temporal.internal.payload.storage.ExternalStorageRunner; -import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.context.SerializationContext; import io.temporal.payload.storage.StorageDriverTargetInfo; import io.temporal.serviceclient.WorkflowServiceStubs; import javax.annotation.Nonnull; @@ -36,12 +36,12 @@ static ManualActivityCompletionClientFactory newFactory( ManualActivityCompletionClient getClient( @Nonnull byte[] taskToken, @Nonnull Scope metricsScope, - @Nullable ActivitySerializationContext activitySerializationContext); + @Nullable SerializationContext serializationContext); ManualActivityCompletionClient getClient( @Nonnull byte[] taskToken, @Nonnull Scope metricsScope, - @Nullable ActivitySerializationContext activitySerializationContext, + @Nullable SerializationContext serializationContext, @Nullable StorageDriverTargetInfo storageTarget); ManualActivityCompletionClient getClient( @@ -53,5 +53,5 @@ ManualActivityCompletionClient getClient( @Nonnull WorkflowExecution execution, @Nonnull String activityId, @Nonnull Scope metricsScope, - @Nullable ActivitySerializationContext activitySerializationContext); + @Nullable SerializationContext serializationContext); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactoryImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactoryImpl.java index 286d11902a..379e9ef867 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactoryImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactoryImpl.java @@ -9,6 +9,7 @@ import io.temporal.internal.payload.storage.ActivityStorageTargets; import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.context.SerializationContext; import io.temporal.payload.storage.StorageDriverTargetInfo; import io.temporal.serviceclient.WorkflowServiceStubs; import java.util.Objects; @@ -45,7 +46,12 @@ public ManualActivityCompletionClient getClient( public ManualActivityCompletionClient getClient( @Nonnull byte[] taskToken, @Nonnull Scope metricsScope, - @Nullable ActivitySerializationContext activitySerializationContext) { + @Nullable SerializationContext serializationContext) { + // Only an activity context identifies the activity and its workflow for the storage target. + ActivitySerializationContext activitySerializationContext = + serializationContext instanceof ActivitySerializationContext + ? (ActivitySerializationContext) serializationContext + : null; StorageDriverTargetInfo storageTarget = activitySerializationContext == null ? null @@ -57,14 +63,14 @@ public ManualActivityCompletionClient getClient( null, Strings.emptyToNull(activitySerializationContext.getWorkflowType())) .build(); - return getClient(taskToken, metricsScope, activitySerializationContext, storageTarget); + return getClient(taskToken, metricsScope, serializationContext, storageTarget); } @Override public ManualActivityCompletionClient getClient( @Nonnull byte[] taskToken, @Nonnull Scope metricsScope, - @Nullable ActivitySerializationContext activitySerializationContext, + @Nullable SerializationContext serializationContext, @Nullable StorageDriverTargetInfo storageTarget) { Preconditions.checkNotNull(metricsScope, "metricsScope"); Preconditions.checkNotNull(taskToken, "taskToken"); @@ -78,7 +84,7 @@ public ManualActivityCompletionClient getClient( taskToken, null, null, - activitySerializationContext, + serializationContext, storageTarget, externalStorage); } @@ -96,10 +102,14 @@ public ManualActivityCompletionClient getClient( @Nonnull WorkflowExecution execution, @Nonnull String activityId, @Nonnull Scope metricsScope, - @Nullable ActivitySerializationContext activitySerializationContext) { + @Nullable SerializationContext serializationContext) { Preconditions.checkNotNull(metricsScope, "metricsScope"); Preconditions.checkNotNull(execution, "execution"); Preconditions.checkNotNull(activityId, "activityId"); + ActivitySerializationContext activitySerializationContext = + serializationContext instanceof ActivitySerializationContext + ? (ActivitySerializationContext) serializationContext + : null; String activityRunId = execution.getWorkflowId().isEmpty() ? Strings.emptyToNull(execution.getRunId()) : null; String activityType = @@ -119,7 +129,7 @@ public ManualActivityCompletionClient getClient( null, execution, activityId, - activitySerializationContext, + serializationContext, ActivityStorageTargets.newBuilder(namespace) .setActivity(activityId, activityRunId, activityType) .setWorkflow( diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java index 5e597cbb58..396f440d69 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java @@ -20,7 +20,7 @@ import io.temporal.internal.common.OptionsUtils; import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.internal.retryer.GrpcRetryer; -import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.context.SerializationContext; import io.temporal.payload.storage.StorageDriverTargetInfo; import io.temporal.serviceclient.RpcRetryOptions; import io.temporal.serviceclient.WorkflowServiceStubs; @@ -57,7 +57,7 @@ class ManualActivityCompletionClientImpl implements ManualActivityCompletionClie @Nullable byte[] taskToken, @Nullable WorkflowExecution execution, @Nullable String activityId, - @Nullable ActivitySerializationContext context, + @Nullable SerializationContext context, @Nullable StorageDriverTargetInfo storageTarget, @Nullable ExternalStorageRunner externalStorage) { this.service = service; diff --git a/temporal-sdk/src/main/java/io/temporal/internal/nexus/PropagatedNexusSerializationContext.java b/temporal-sdk/src/main/java/io/temporal/internal/nexus/PropagatedNexusSerializationContext.java new file mode 100644 index 0000000000..9690a6a925 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/nexus/PropagatedNexusSerializationContext.java @@ -0,0 +1,49 @@ +package io.temporal.internal.nexus; + +import io.temporal.api.nexus.v1.PropagatedSerializationContext; +import io.temporal.payload.context.NexusSerializationContext; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; + +/** + * Converts between {@link NexusSerializationContext} and the {@link PropagatedSerializationContext} + * the server stores on executions started by a Nexus operation handler. + * + *

When a Nexus handler starts a workflow or a standalone activity, the SDK sends the handler's + * serialization context with the start request. The server stores it and returns it to workers and + * to clients reading the execution's result, so the result and failure are encoded and decoded with + * the same context. See {@link NexusSerializationContext} for which payloads use it. + */ +public final class PropagatedNexusSerializationContext { + + /** + * The serialization context of the Nexus operation being handled on the current thread, in the + * form sent with start requests, or {@code null} if there is no Nexus task in scope or its + * service and operation are not known yet. + */ + public static @Nullable PropagatedSerializationContext current() { + if (!CurrentNexusOperationContext.isNexusContext()) { + return null; + } + NexusSerializationContext context = + CurrentNexusOperationContext.get().getSerializationContext(); + return context != null ? toProto(context) : null; + } + + public static @Nonnull PropagatedSerializationContext toProto( + @Nonnull NexusSerializationContext context) { + return PropagatedSerializationContext.newBuilder() + .setEndpoint(context.getEndpoint()) + .setService(context.getService()) + .setOperation(context.getOperation()) + .build(); + } + + public static @Nonnull NexusSerializationContext fromProto( + @Nonnull PropagatedSerializationContext context) { + return new NexusSerializationContext( + context.getEndpoint(), context.getService(), context.getOperation()); + } + + private PropagatedNexusSerializationContext() {} +} diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/DynamicSyncWorkflowDefinition.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/DynamicSyncWorkflowDefinition.java index 7fd9fc00be..ea9a4d96a0 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/DynamicSyncWorkflowDefinition.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/DynamicSyncWorkflowDefinition.java @@ -53,7 +53,11 @@ public Optional execute(Header header, Optional input) { WorkflowInboundCallsInterceptor.WorkflowOutput result = workflowInvoker.execute( new WorkflowInboundCallsInterceptor.WorkflowInput(header, new Object[] {args})); - return dataConverterWithWorkflowContext.toPayloads(result.getResult()); + // The result of a workflow started by a Nexus operation handler is returned to the Nexus + // caller, so it is encoded with the handler's Nexus context. + return WorkflowInternal.getRootWorkflowContext() + .getDataConverterForResult() + .toPayloads(result.getResult()); } @Nullable diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/POJOWorkflowImplementationFactory.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/POJOWorkflowImplementationFactory.java index fb20638296..f69d416b42 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/POJOWorkflowImplementationFactory.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/POJOWorkflowImplementationFactory.java @@ -362,7 +362,11 @@ public Optional execute(Header header, Optional input) if (workflowMethod.getReturnType() == Void.TYPE) { return Optional.empty(); } - return dataConverterWithWorkflowContext.toPayloads(result.getResult()); + // The result of a workflow started by a Nexus operation handler is returned to the Nexus + // caller, so it is encoded with the handler's Nexus context. + return WorkflowInternal.getRootWorkflowContext() + .getDataConverterForResult() + .toPayloads(result.getResult()); } @Nullable diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflow.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflow.java index 99ffbcbd0c..d8c7cc33fe 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflow.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflow.java @@ -12,6 +12,7 @@ import io.temporal.common.converter.RawValue; import io.temporal.internal.logging.LoggerTag; import io.temporal.internal.logging.PrefixedMdc; +import io.temporal.internal.nexus.PropagatedNexusSerializationContext; import io.temporal.internal.replay.ReplayWorkflow; import io.temporal.internal.replay.ReplayWorkflowContext; import io.temporal.internal.replay.WorkflowContext; @@ -104,6 +105,11 @@ public void start(HistoryEvent event, ReplayWorkflowContext context) { } this.workflowContext.setReplayContext(context); + this.workflowContext.setPropagatedNexusSerializationContext( + startEvent.hasPropagatedNexusSerializationContext() + ? PropagatedNexusSerializationContext.fromProto( + startEvent.getPropagatedNexusSerializationContext()) + : null); workflowProc = new WorkflowExecutionHandler( diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java index 2d5cfcba33..3f1932f42c 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java @@ -104,6 +104,10 @@ final class SyncWorkflowContext implements WorkflowContext, WorkflowOutboundCall private boolean readOnly = false; private final WorkflowThreadLocal currentUpdateInfo = new WorkflowThreadLocal<>(); @Nullable private String currentDetails; + // Set when the workflow was started by a Nexus operation handler. Its result and the failure that + // closes it are returned to the Nexus caller, so they are serialized with the handler's Nexus + // context instead of the workflow context. Everything else keeps the workflow context. + @Nullable private DataConverter dataConverterWithPropagatedNexusContext; public SyncWorkflowContext( @Nonnull String namespace, @@ -1300,6 +1304,30 @@ public DataConverter getDataConverterWithCurrentWorkflowContext() { return dataConverterWithCurrentWorkflowContext; } + /** + * Records the serialization context of the Nexus operation handler that started this workflow, as + * propagated by the server in the WorkflowExecutionStarted event. + */ + void setPropagatedNexusSerializationContext( + @Nullable NexusSerializationContext propagatedNexusSerializationContext) { + this.dataConverterWithPropagatedNexusContext = + propagatedNexusSerializationContext == null + ? null + : dataConverter.withContext(propagatedNexusSerializationContext); + } + + /** + * The data converter for the workflow result and the failure that closes the workflow, which are + * returned to the Nexus caller: the propagated Nexus context if the workflow was started by a + * Nexus operation handler, otherwise the workflow context. Input, memo, signals, queries and + * updates always use the workflow context. + */ + DataConverter getDataConverterForResult() { + return dataConverterWithPropagatedNexusContext != null + ? dataConverterWithPropagatedNexusContext + : dataConverterWithCurrentWorkflowContext; + } + boolean isReplaying() { return replayContext.isReplaying(); } @@ -1625,6 +1653,15 @@ public WorkflowImplementationOptions getWorkflowImplementationOptions() { @Override public Failure mapWorkflowExceptionToFailure(Throwable failure) { + return getDataConverterForResult().exceptionToFailure(failure); + } + + /** + * Converts an exception thrown by an update validator or handler into the failure that rejects or + * fails the update. Unlike {@link #mapWorkflowExceptionToFailure}, this always uses the workflow + * context, because the failure belongs to the update rather than closing the workflow. + */ + Failure mapUpdateExceptionToFailure(Throwable failure) { return dataConverterWithCurrentWorkflowContext.exceptionToFailure(failure); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowExecutionHandler.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowExecutionHandler.java index afeb70b14b..5ab392bc7b 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowExecutionHandler.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowExecutionHandler.java @@ -105,7 +105,7 @@ public void handleValidateUpdate( try { context.handleValidateUpdate(updateName, updateId, input, eventId, new Header(header)); } catch (Throwable e) { - applyWorkflowFailurePolicyAndRethrow(e); + applyWorkflowFailurePolicyAndRethrow(e, true); } } @@ -121,12 +121,20 @@ public Optional handleExecuteUpdate( // Re-throw to fail the workflow task throw e; } catch (Throwable e) { - applyWorkflowFailurePolicyAndRethrow(e); + applyWorkflowFailurePolicyAndRethrow(e, true); } return Optional.empty(); } private void applyWorkflowFailurePolicyAndRethrow(Throwable e) { + applyWorkflowFailurePolicyAndRethrow(e, false); + } + + /** + * @param forUpdate whether the exception was thrown by an update validator or handler, in which + * case the resulting failure rejects or fails the update rather than closing the workflow + */ + private void applyWorkflowFailurePolicyAndRethrow(Throwable e, boolean forUpdate) { if (e instanceof DestroyWorkflowThreadError) { throw (DestroyWorkflowThreadError) e; } @@ -134,18 +142,18 @@ private void applyWorkflowFailurePolicyAndRethrow(Throwable e) { Class[] failTypes = implementationOptions.getFailWorkflowExceptionTypes(); if (exception instanceof TemporalFailure) { - throwAndFailWorkflowExecution(exception); + throwAndFailWorkflowExecution(exception, forUpdate); } for (Class failType : failTypes) { if (failType.isAssignableFrom(exception.getClass())) { - throwAndFailWorkflowExecution(exception); + throwAndFailWorkflowExecution(exception, forUpdate); } } throw wrap(exception); } - private void throwAndFailWorkflowExecution(Throwable exception) { + private void throwAndFailWorkflowExecution(Throwable exception, boolean forUpdate) { ReplayWorkflowContext replayWorkflowContext = context.getReplayContext(); @Nullable String fullReplayDirectQueryName = replayWorkflowContext.getFullReplayDirectQueryName(); @@ -174,7 +182,12 @@ private void throwAndFailWorkflowExecution(Throwable exception) { } } - throw new WorkflowExecutionException(context.mapWorkflowExceptionToFailure(exception)); + // An update failure belongs to the update, so it keeps the workflow context even for a workflow + // started by a Nexus operation handler, whose closing failure uses the handler's Nexus context. + throw new WorkflowExecutionException( + forUpdate + ? context.mapUpdateExceptionToFailure(exception) + : context.mapWorkflowExceptionToFailure(exception)); } /** diff --git a/temporal-sdk/src/main/java/io/temporal/payload/context/NexusSerializationContext.java b/temporal-sdk/src/main/java/io/temporal/payload/context/NexusSerializationContext.java index 3e355c1a14..8c46e23fd3 100644 --- a/temporal-sdk/src/main/java/io/temporal/payload/context/NexusSerializationContext.java +++ b/temporal-sdk/src/main/java/io/temporal/payload/context/NexusSerializationContext.java @@ -9,22 +9,19 @@ * {@link SerializationContext} for Nexus operation payloads, identifying the Nexus endpoint, * service, and resolved operation the payload belongs to. * - *

Callers receive this context when encoding operation inputs and when decoding operation - * results and failures. Handlers receive it when decoding operation inputs, encoding synchronous - * operation results, and encoding failures produced while handling a Nexus task. + *

This context is used for payloads exchanged between a Nexus caller and handler: operation + * inputs, results, and failures. When a handler starts a workflow or standalone activity to back an + * operation, whatever it returns to the caller, such as its result or failure, is serialized with + * this context too, so the caller can decode it. Everything else keeps the workflow's or activity's + * own context. This requires a server that propagates the context. * - *

The context is not propagated to the eventual result of an asynchronous operation, because the - * operation is completed out of band rather than by the task the handler was invoked for. A - * standalone operation handle uses the context of its start request, including when the start - * request returns an already-running operation; a handle obtained by operation ID without starting - * an operation has no endpoint, service, or operation to build a context from and therefore - * serializes without one. + *

To complete such an activity asynchronously, store the context from {@link + * io.temporal.activity.ActivityInfo#getNexusSerializationContext()} with the task token and pass it + * to {@link io.temporal.client.ActivityCompletionClient#withContext(NexusSerializationContext)}. * - *

Failure conversion is not symmetric: a failure is encoded by the handler and decoded by the - * caller, so an implementation sees this context on only one side of a given failure, and for some - * operation paths it sees no context at all. Context-dependent encodings must therefore be - * self-describing, and decoders must keep accepting payloads that were encoded without a context. - * This applies to {@link FailureConverter} as much as to payload encoding. + *

Some payloads, such as failures and payloads from older servers, may be serialized without + * this context. Context-dependent encodings, including in a {@link FailureConverter}, should be + * self-describing, and decoders should accept payloads encoded without a context. */ @Experimental public final class NexusSerializationContext implements SerializationContext { diff --git a/temporal-sdk/src/test/java/io/temporal/client/ActivityCompletionClientImplTest.java b/temporal-sdk/src/test/java/io/temporal/client/ActivityCompletionClientImplTest.java index 12f45a7745..c3b8abe0b1 100644 --- a/temporal-sdk/src/test/java/io/temporal/client/ActivityCompletionClientImplTest.java +++ b/temporal-sdk/src/test/java/io/temporal/client/ActivityCompletionClientImplTest.java @@ -1,5 +1,6 @@ package io.temporal.client; +import static org.junit.Assert.assertEquals; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.*; @@ -8,12 +9,20 @@ import com.uber.m3.tally.Scope; import io.temporal.activity.ManualActivityCompletionClient; import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.workflowservice.v1.GetSystemInfoResponse; +import io.temporal.api.workflowservice.v1.RespondActivityTaskCompletedRequest; +import io.temporal.api.workflowservice.v1.WorkflowServiceGrpc; import io.temporal.internal.client.external.ManualActivityCompletionClientFactory; +import io.temporal.internal.nexus.ContextRecordingCodec; import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.context.NexusSerializationContext; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.serviceclient.WorkflowServiceStubsOptions; import io.temporal.workflow.Functions; import java.util.Optional; import org.junit.Before; import org.junit.Test; +import org.mockito.ArgumentCaptor; public class ActivityCompletionClientImplTest { @@ -186,4 +195,36 @@ public void testHeartbeatStandaloneDoesNotCallCompletionHandle() client.heartbeatStandalone("my-act", Optional.empty(), null); verify(completionHandle, never()).apply(); } + + @Test + public void taskTokenCompletionUsesNexusContext() { + WorkflowServiceStubs service = mock(WorkflowServiceStubs.class); + WorkflowServiceGrpc.WorkflowServiceBlockingStub blockingStub = + mock(WorkflowServiceGrpc.WorkflowServiceBlockingStub.class); + when(service.getServerCapabilities()) + .thenReturn(() -> GetSystemInfoResponse.Capabilities.getDefaultInstance()); + when(service.getOptions()).thenReturn(WorkflowServiceStubsOptions.getDefaultInstance()); + when(service.blockingStub()).thenReturn(blockingStub); + when(blockingStub.withOption(any(), any())).thenReturn(blockingStub); + ContextRecordingCodec codec = new ContextRecordingCodec(); + NexusSerializationContext nexusContext = + new NexusSerializationContext("endpoint", "service", "operation"); + ActivityCompletionClient completionClient = + new ActivityCompletionClientImpl( + ManualActivityCompletionClientFactory.newFactory( + service, "test-namespace", "test-identity", codec.dataConverter()), + completionHandle, + scope, + null); + + completionClient.withContext(nexusContext).complete(new byte[] {1, 2, 3}, "result"); + + // The result goes back to the Nexus caller, so it's encoded with the Nexus context. + ArgumentCaptor request = + ArgumentCaptor.forClass(RespondActivityTaskCompletedRequest.class); + verify(blockingStub).respondActivityTaskCompleted(request.capture()); + assertEquals( + ContextRecordingCodec.describe(nexusContext), + ContextRecordingCodec.contextOf(request.getValue().getResult().getPayloads(0))); + } } diff --git a/temporal-sdk/src/test/java/io/temporal/client/nexus/StandaloneNexusSerializationContextTest.java b/temporal-sdk/src/test/java/io/temporal/client/nexus/StandaloneNexusSerializationContextTest.java index e8a100c2f6..eff39333cc 100644 --- a/temporal-sdk/src/test/java/io/temporal/client/nexus/StandaloneNexusSerializationContextTest.java +++ b/temporal-sdk/src/test/java/io/temporal/client/nexus/StandaloneNexusSerializationContextTest.java @@ -58,9 +58,6 @@ public class StandaloneNexusSerializationContextTest { // contexts it records are the client's. private static final RecordingCodec CODEC = new RecordingCodec(); - // A handle obtained by operation ID decodes a context-encoded payload without a context, so - // that direction is only an error when a test says it should be. - private static boolean allowContextlessDecodeOfSignedPayload; private static final RecordingFailureConverter FAILURE_CONVERTER = new RecordingFailureConverter(); @@ -97,7 +94,6 @@ public void requireStandaloneNexusSupport() { testWorkflowRule.isUseExternalService()); CODEC.reset(); FAILURE_CONVERTER.reset(); - allowContextlessDecodeOfSignedPayload = false; } @Test @@ -187,22 +183,20 @@ public void describeReadsTheSummaryUnderTheOperationsContext() { } @Test - public void handleObtainedByIdHasNoContext() { - allowContextlessDecodeOfSignedPayload = true; + public void handleObtainedByIdUsesContextFromServer() { String input = "ping-" + UUID.randomUUID(); UntypedNexusOperationHandle started = startOperation(input); started.getResult(String.class); CODEC.reset(); - // A handle obtained by ID never saw a start request, so there is no endpoint, service or - // operation to scope its converter by. + // A handle obtained by ID never saw a start request, so it relies on the server returning the + // operation's context with the poll response. UntypedNexusOperationHandle detached = nexusClient().getHandle(started.getNexusOperationId(), started.getNexusOperationRunId()); Assert.assertEquals("echo:" + input, detached.getResult(String.class)); - Assert.assertEquals( - "a handle obtained by operation ID should decode without a Nexus context", - Collections.emptyList(), + "a handle obtained by operation ID should decode with the context the server returns", + Collections.singletonList(expectedContext()), CODEC.nexusContexts()); } @@ -371,20 +365,15 @@ public List decode(@Nonnull List payloads) { decoded.add(payload); continue; } - if (!(context instanceof NexusSerializationContext)) { - // The reverse mismatch: encoded under a context, decoded without one. Expected only - // for a handle obtained by operation ID, which opts in below. - Assert.assertTrue( - "payload encoded under a Nexus context was decoded under " + context, - allowContextlessDecodeOfSignedPayload); - } - if (context instanceof NexusSerializationContext) { - NexusSerializationContext nexus = (NexusSerializationContext) context; - Assert.assertEquals( - "payload should be decoded under the context it was encoded with", - nexus.getEndpoint() + ":" + nexus.getService() + ":" + nexus.getOperation(), - signature.toStringUtf8()); - } + // The reverse mismatch: encoded under a context, decoded without one. + Assert.assertTrue( + "payload encoded under a Nexus context was decoded under " + context, + context instanceof NexusSerializationContext); + NexusSerializationContext nexus = (NexusSerializationContext) context; + Assert.assertEquals( + "payload should be decoded under the context it was encoded with", + nexus.getEndpoint() + ":" + nexus.getService() + ":" + nexus.getOperation(), + signature.toStringUtf8()); decoded.add(Payload.newBuilder(payload).removeMetadata(SIGNATURE_KEY).build()); } return decoded; diff --git a/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityNexusSerializationContextTest.java b/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityNexusSerializationContextTest.java new file mode 100644 index 0000000000..85b6fd5e22 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityNexusSerializationContextTest.java @@ -0,0 +1,270 @@ +package io.temporal.internal.activity; + +import static io.temporal.internal.nexus.ContextRecordingCodec.contextOf; +import static io.temporal.internal.nexus.ContextRecordingCodec.stamp; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +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.NoopScope; +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import io.temporal.api.common.v1.ActivityType; +import io.temporal.api.common.v1.Payloads; +import io.temporal.api.nexus.v1.PropagatedSerializationContext; +import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse; +import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatRequest; +import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatResponse; +import io.temporal.api.workflowservice.v1.WorkflowServiceGrpc; +import io.temporal.client.WorkflowClient; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.interceptors.WorkerInterceptor; +import io.temporal.failure.ApplicationFailure; +import io.temporal.internal.client.external.ManualActivityCompletionClientFactory; +import io.temporal.internal.common.ProtobufTimeUtils; +import io.temporal.internal.nexus.ContextRecordingCodec; +import io.temporal.internal.worker.ActivityTask; +import io.temporal.internal.worker.ActivityTaskHandler; +import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.context.NexusSerializationContext; +import io.temporal.payload.context.SerializationContext; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.worker.tuning.SlotPermit; +import java.time.Duration; +import java.time.Instant; +import java.util.Collections; +import java.util.Optional; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.mockito.ArgumentCaptor; + +/** + * Unit tests that a standalone activity started by a Nexus operation handler encodes its result, + * failure and heartbeat details with the handler's Nexus context, as propagated by the server, + * while its input keeps the activity context. + */ +public class ActivityNexusSerializationContextTest { + + private static final String NAMESPACE = "test-namespace"; + private static final String TASK_QUEUE = "tq"; + private static final NexusSerializationContext NEXUS_CONTEXT = + new NexusSerializationContext("endpoint", "service", "operation"); + private static final PropagatedSerializationContext PROPAGATED = + PropagatedSerializationContext.newBuilder() + .setEndpoint("endpoint") + .setService("service") + .setOperation("operation") + .build(); + private static final String NEXUS = ContextRecordingCodec.describe(NEXUS_CONTEXT); + private static final String ACTIVITY = "activity:Echo"; + + @ActivityInterface + public interface TestActivity { + @ActivityMethod(name = "Echo") + String echo(String input); + } + + public static class TestActivityImpl implements TestActivity { + @Override + public String echo(String input) { + if ("fail".equals(input)) { + throw ApplicationFailure.newNonRetryableFailure("failed", "TestFailure", "detail"); + } + return input; + } + } + + private ContextRecordingCodec codec; + private ActivityTaskHandlerImpl handler; + + @Before + public void setUp() { + codec = new ContextRecordingCodec(); + ActivityExecutionContextFactory contextFactory = mock(ActivityExecutionContextFactory.class); + when(contextFactory.createContext(any(), any(), any())) + .thenAnswer( + invocation -> { + InternalActivityExecutionContext context = + mock(InternalActivityExecutionContext.class); + when(context.getInfo()).thenReturn(invocation.getArgument(0)); + when(context.getLastHeartbeatValue()).thenReturn("last-heartbeat"); + return context; + }); + handler = + new ActivityTaskHandlerImpl( + NAMESPACE, + TASK_QUEUE, + codec.dataConverter(), + contextFactory, + new WorkerInterceptor[0], + Collections.emptyList()); + handler.registerActivityImplementations(new Object[] {new TestActivityImpl()}); + } + + @Test + public void propagatedContextIsUsedOnlyForResult() { + ActivityTaskHandler.Result result = handle("hello", PROPAGATED); + + Assert.assertEquals(Collections.singletonList(ACTIVITY), codec.decodeContexts()); + Assert.assertEquals(NEXUS, contextOf(result.getTaskCompleted().getResult().getPayloads(0))); + } + + @Test + public void propagatedContextIsUsedForFailureAndHeartbeatDetails() { + ActivityTaskHandler.Result result = handle("fail", PROPAGATED); + + Assert.assertEquals( + NEXUS, + contextOf( + result + .getTaskFailed() + .getTaskFailedRequest() + .getFailure() + .getApplicationFailureInfo() + .getDetails() + .getPayloads(0))); + Assert.assertEquals( + NEXUS, + contextOf( + result + .getTaskFailed() + .getTaskFailedRequest() + .getLastHeartbeatDetails() + .getPayloads(0))); + } + + @Test + public void withoutPropagatedContextActivityContextIsUsed() { + ActivityTaskHandler.Result result = handle("hello", null); + + Assert.assertEquals(Collections.singletonList(ACTIVITY), codec.decodeContexts()); + Assert.assertEquals(ACTIVITY, contextOf(result.getTaskCompleted().getResult().getPayloads(0))); + } + + @Test + public void propagatedContextIsUsedForHeartbeats() { + WorkflowServiceStubs service = mock(WorkflowServiceStubs.class); + WorkflowServiceGrpc.WorkflowServiceBlockingStub blockingStub = + mock(WorkflowServiceGrpc.WorkflowServiceBlockingStub.class); + when(service.blockingStub()).thenReturn(blockingStub); + when(blockingStub.withOption(any(), any())).thenReturn(blockingStub); + when(blockingStub.recordActivityTaskHeartbeat(any())) + .thenReturn(RecordActivityTaskHeartbeatResponse.getDefaultInstance()); + ActivityInfoInternal info = mock(ActivityInfoInternal.class); + when(info.getTaskToken()).thenReturn(new byte[] {1}); + when(info.getActivityType()).thenReturn("Echo"); + when(info.getHeartbeatTimeout()).thenReturn(Duration.ZERO); + when(info.getNexusSerializationContext()).thenReturn(NEXUS_CONTEXT); + when(info.getHeartbeatDetails()) + .thenReturn( + Optional.of( + Payloads.newBuilder() + .addPayloads( + stamp( + DefaultDataConverter.STANDARD_INSTANCE.toPayload("previous").get(), + NEXUS_CONTEXT)) + .build())); + ScheduledExecutorService heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(); + try { + HeartbeatContextImpl heartbeatContext = + new HeartbeatContextImpl( + service, + NAMESPACE, + info, + codec.dataConverter(), + heartbeatExecutor, + new NoopScope(), + "test-identity", + Duration.ofSeconds(60), + Duration.ofSeconds(30), + null); + + // The previous attempt's details are decoded, and new details encoded, with the Nexus + // context, because the server puts them into the timeout failure returned to the caller. + Assert.assertEquals( + Optional.of("previous"), + heartbeatContext.getHeartbeatDetails(String.class, String.class)); + Assert.assertEquals(Collections.singletonList(NEXUS), codec.decodeContexts()); + heartbeatContext.heartbeat("details"); + ArgumentCaptor request = + ArgumentCaptor.forClass(RecordActivityTaskHeartbeatRequest.class); + verify(blockingStub).recordActivityTaskHeartbeat(request.capture()); + Assert.assertEquals(NEXUS, contextOf(request.getValue().getDetails().getPayloads(0))); + } finally { + heartbeatExecutor.shutdownNow(); + } + } + + @Test + public void propagatedContextIsUsedForLocalManualCompletion() { + WorkflowClient client = mock(WorkflowClient.class); + when(client.getWorkflowServiceStubs()).thenReturn(mock(WorkflowServiceStubs.class)); + ActivityInfoInternal info = mock(ActivityInfoInternal.class); + when(info.getNamespace()).thenReturn(NAMESPACE); + when(info.getActivityType()).thenReturn("Echo"); + when(info.getTaskToken()).thenReturn(new byte[] {1}); + when(info.getNexusSerializationContext()).thenReturn(NEXUS_CONTEXT); + ManualActivityCompletionClientFactory completionClientFactory = + mock(ManualActivityCompletionClientFactory.class); + ActivityExecutionContextImpl context = + new ActivityExecutionContextImpl( + client, + NAMESPACE, + new Object(), + info, + codec.dataConverter(), + mock(ScheduledExecutorService.class), + completionClientFactory, + () -> {}, + new NoopScope(), + "test-identity", + Duration.ofSeconds(60), + Duration.ofSeconds(30), + () -> {}, + null); + + context.useLocalManualCompletion(); + + // The manual completion client reports the result, failure and heartbeats, so it uses the + // Nexus context too. + verify(completionClientFactory).getClient(any(), any(), eq(NEXUS_CONTEXT), any()); + } + + private ActivityTaskHandler.Result handle( + String input, PropagatedSerializationContext propagated) { + return handler.handle( + new ActivityTask(pollResponse(input, propagated), new SlotPermit(), () -> {}), + new NoopScope(), + false); + } + + private static PollActivityTaskQueueResponse pollResponse( + String input, PropagatedSerializationContext propagated) { + // The input is encoded with the activity context whether or not a context was propagated. + SerializationContext encodedWith = + new ActivitySerializationContext(NAMESPACE, null, null, "Echo", TASK_QUEUE, false); + PollActivityTaskQueueResponse.Builder response = + PollActivityTaskQueueResponse.newBuilder() + .setTaskToken(ByteString.copyFromUtf8("token")) + .setActivityId("activity-id") + .setActivityType(ActivityType.newBuilder().setName("Echo")) + .setScheduledTime(ProtobufTimeUtils.toProtoTimestamp(Instant.EPOCH)) + .setStartedTime(ProtobufTimeUtils.toProtoTimestamp(Instant.EPOCH)) + .setInput( + Payloads.newBuilder() + .addPayloads( + stamp( + DefaultDataConverter.STANDARD_INSTANCE.toPayload(input).get(), + encodedWith))); + if (propagated != null) { + response.setPropagatedNexusSerializationContext(propagated); + } + return response.build(); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerNexusSerializationContextTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerNexusSerializationContextTest.java new file mode 100644 index 0000000000..2ca127cfa9 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerNexusSerializationContextTest.java @@ -0,0 +1,290 @@ +package io.temporal.internal.client; + +import static io.temporal.internal.nexus.ContextRecordingCodec.contextOf; +import static io.temporal.internal.nexus.ContextRecordingCodec.stamp; +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 com.uber.m3.tally.NoopScope; +import io.grpc.Deadline; +import io.temporal.api.activity.v1.ActivityExecutionInfo; +import io.temporal.api.activity.v1.ActivityExecutionOutcome; +import io.temporal.api.common.v1.ActivityType; +import io.temporal.api.common.v1.Payloads; +import io.temporal.api.failure.v1.ApplicationFailureInfo; +import io.temporal.api.failure.v1.Failure; +import io.temporal.api.nexus.v1.PropagatedSerializationContext; +import io.temporal.api.sdk.v1.UserMetadata; +import io.temporal.api.workflowservice.v1.DescribeActivityExecutionResponse; +import io.temporal.api.workflowservice.v1.PollActivityExecutionRequest; +import io.temporal.api.workflowservice.v1.PollActivityExecutionResponse; +import io.temporal.api.workflowservice.v1.StartActivityExecutionRequest; +import io.temporal.api.workflowservice.v1.StartActivityExecutionResponse; +import io.temporal.client.ActivityExecutionDescription; +import io.temporal.client.ActivityFailedException; +import io.temporal.client.StartActivityOptions; +import io.temporal.client.WorkflowClient; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.interceptors.ActivityClientCallsInterceptor.GetActivityResultInput; +import io.temporal.common.interceptors.ActivityClientCallsInterceptor.StartActivityInput; +import io.temporal.common.interceptors.Header; +import io.temporal.internal.client.external.GenericWorkflowClient; +import io.temporal.internal.nexus.ContextRecordingCodec; +import io.temporal.internal.nexus.CurrentNexusOperationContext; +import io.temporal.internal.nexus.InternalNexusOperationContext; +import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.context.NexusSerializationContext; +import io.temporal.payload.context.SerializationContext; +import java.time.Duration; +import java.util.Arrays; +import java.util.Collections; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.mockito.ArgumentCaptor; + +/** + * Unit tests for propagating the Nexus serialization context to standalone activities started by a + * Nexus operation handler, and for decoding their results with the context the server returns. + */ +public class RootActivityClientInvokerNexusSerializationContextTest { + + private static final String NAMESPACE = "test-namespace"; + private static final String ACTIVITY_TYPE = "TestActivity"; + private static final NexusSerializationContext NEXUS_CONTEXT = + new NexusSerializationContext("endpoint", "service", "operation"); + private static final PropagatedSerializationContext PROPAGATED = + PropagatedSerializationContext.newBuilder() + .setEndpoint("endpoint") + .setService("service") + .setOperation("operation") + .build(); + private static final String NEXUS = ContextRecordingCodec.describe(NEXUS_CONTEXT); + + private ContextRecordingCodec codec; + private GenericWorkflowClient genericClient; + private RootActivityClientInvoker invoker; + + @Before + public void setUp() { + codec = new ContextRecordingCodec(); + genericClient = mock(GenericWorkflowClient.class); + invoker = + new RootActivityClientInvoker( + genericClient, NAMESPACE, "test-identity", codec.dataConverter()); + } + + @After + public void tearDown() { + CurrentNexusOperationContext.unset(); + } + + @Test + public void startFromNexusHandlerPropagatesContextButEncodesInputWithActivityContext() { + InternalNexusOperationContext nexusContext = + new InternalNexusOperationContext( + NAMESPACE, "tq", "endpoint", new NoopScope(), mock(WorkflowClient.class)); + nexusContext.setSerializationContext(NEXUS_CONTEXT); + CurrentNexusOperationContext.set(nexusContext); + when(genericClient.startActivity(any())) + .thenReturn(StartActivityExecutionResponse.getDefaultInstance()); + + invoker.startActivity(newStartInput()); + + // Only the result and failure use the propagated context; the input and user metadata stay + // within the activity. + StartActivityExecutionRequest sent = captureStartRequest(); + Assert.assertEquals(PROPAGATED, sent.getPropagatedNexusSerializationContext()); + Assert.assertEquals("activity:" + ACTIVITY_TYPE, contextOf(sent.getInput().getPayloads(0))); + Assert.assertEquals( + "activity:" + ACTIVITY_TYPE, contextOf(sent.getUserMetadata().getSummary())); + } + + @Test + public void startOutsideNexusHandlerDoesNotPropagateContext() { + when(genericClient.startActivity(any())) + .thenReturn(StartActivityExecutionResponse.getDefaultInstance()); + + invoker.startActivity(newStartInput()); + + StartActivityExecutionRequest sent = captureStartRequest(); + Assert.assertFalse(sent.hasPropagatedNexusSerializationContext()); + Assert.assertEquals("activity:" + ACTIVITY_TYPE, contextOf(sent.getInput().getPayloads(0))); + } + + @Test + public void getResultDecodesWithPropagatedContext() throws Exception { + when(genericClient.pollActivity(any(PollActivityExecutionRequest.class), any(Deadline.class))) + .thenReturn(resultResponse(PROPAGATED)); + + String result = invoker.getActivityResult(newGetResultInput()).getResult(); + + Assert.assertEquals("result", result); + Assert.assertEquals(Collections.singletonList(NEXUS), codec.decodeContexts()); + } + + @Test + public void getResultWithoutPropagatedContextUsesActivityContext() throws Exception { + when(genericClient.pollActivity(any(PollActivityExecutionRequest.class), any(Deadline.class))) + .thenReturn(resultResponse(null)); + + invoker.getActivityResult(newGetResultInput()); + + // The result path doesn't know the activity type, so the activity context carries none. + Assert.assertEquals(Collections.singletonList("activity:null"), codec.decodeContexts()); + } + + @Test + public void getResultDecodesFailureWithPropagatedContext() { + when(genericClient.pollActivity(any(PollActivityExecutionRequest.class), any(Deadline.class))) + .thenReturn(failureResponse(PROPAGATED)); + + Assert.assertThrows( + ActivityFailedException.class, () -> invoker.getActivityResult(newGetResultInput())); + Assert.assertEquals(Collections.singletonList(NEXUS), codec.decodeContexts()); + } + + @Test + public void getResultAsyncDecodesWithPropagatedContext() throws Exception { + when(genericClient.pollActivityAsync( + any(PollActivityExecutionRequest.class), any(Deadline.class))) + .thenReturn(CompletableFuture.completedFuture(resultResponse(PROPAGATED))); + + String result = invoker.getActivityResultAsync(newGetResultInput()).get().getResult(); + + Assert.assertEquals("result", result); + Assert.assertEquals(Collections.singletonList(NEXUS), codec.decodeContexts()); + } + + @Test + public void describeDecodesResultFailuresAndHeartbeatDetailsWithPropagatedContext() { + ActivityExecutionDescription description = + new ActivityExecutionDescription( + describeResponse(PROPAGATED), codec.dataConverter(), NAMESPACE); + + Assert.assertEquals("result", description.getResult(String.class).get()); + Assert.assertNotNull(description.getLastFailure()); + Assert.assertEquals("heartbeat", description.getHeartbeatDetails().get(0, String.class)); + Assert.assertEquals(Arrays.asList(NEXUS, NEXUS, NEXUS), codec.decodeContexts()); + + codec.reset(); + Assert.assertEquals("input", description.getInput().get(0, String.class)); + Assert.assertEquals("summary", description.getSummary()); + String activity = "activity:" + ACTIVITY_TYPE; + Assert.assertEquals(Arrays.asList(activity, activity), codec.decodeContexts()); + } + + @Test + public void describeWithoutPropagatedContextUsesActivityContext() { + ActivityExecutionDescription description = + new ActivityExecutionDescription(describeResponse(null), codec.dataConverter(), NAMESPACE); + + description.getResult(String.class); + Assert.assertEquals( + Collections.singletonList("activity:" + ACTIVITY_TYPE), codec.decodeContexts()); + } + + private StartActivityExecutionRequest captureStartRequest() { + ArgumentCaptor captor = + ArgumentCaptor.forClass(StartActivityExecutionRequest.class); + verify(genericClient).startActivity(captor.capture()); + return captor.getValue(); + } + + private static StartActivityInput newStartInput() { + StartActivityOptions options = + StartActivityOptions.newBuilder() + .setId("activity-id") + .setTaskQueue("tq") + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .setSummary("summary") + .build(); + return new StartActivityInput( + ACTIVITY_TYPE, Collections.singletonList("input"), options, Header.empty()); + } + + private static GetActivityResultInput newGetResultInput() { + return new GetActivityResultInput<>( + "activity-id", null, String.class, String.class, 10, TimeUnit.SECONDS); + } + + private static PollActivityExecutionResponse resultResponse( + PropagatedSerializationContext propagated) { + PollActivityExecutionResponse.Builder response = + PollActivityExecutionResponse.newBuilder() + .setOutcome( + ActivityExecutionOutcome.newBuilder().setResult(payloads("result", propagated))); + if (propagated != null) { + response.setPropagatedNexusSerializationContext(propagated); + } + return response.build(); + } + + private static PollActivityExecutionResponse failureResponse( + PropagatedSerializationContext propagated) { + Failure failure = + Failure.newBuilder() + .setMessage("failed") + .setApplicationFailureInfo( + ApplicationFailureInfo.newBuilder() + .setType("TestFailure") + .setDetails(payloads("detail", propagated))) + .build(); + PollActivityExecutionResponse.Builder response = + PollActivityExecutionResponse.newBuilder() + .setOutcome(ActivityExecutionOutcome.newBuilder().setFailure(failure)); + if (propagated != null) { + response.setPropagatedNexusSerializationContext(propagated); + } + return response.build(); + } + + private static DescribeActivityExecutionResponse describeResponse( + PropagatedSerializationContext propagated) { + ActivityExecutionInfo.Builder info = + ActivityExecutionInfo.newBuilder() + .setActivityId("activity-id") + .setActivityType(ActivityType.newBuilder().setName(ACTIVITY_TYPE)) + .setTaskQueue("tq"); + // The result, failures and heartbeat details are encoded with the propagated context; the + // input and user metadata keep the activity context. + info.setHeartbeatDetails(payloads("heartbeat", propagated)) + .setUserMetadata( + UserMetadata.newBuilder().setSummary(payloads("summary", null).getPayloads(0))) + .setLastFailure( + Failure.newBuilder() + .setMessage("attempt failed") + .setApplicationFailureInfo( + ApplicationFailureInfo.newBuilder() + .setType("TestFailure") + .setDetails(payloads("detail", propagated)))); + if (propagated != null) { + info.setPropagatedNexusSerializationContext(propagated); + } + return DescribeActivityExecutionResponse.newBuilder() + .setInfo(info) + .setInput(payloads("input", null)) + .setOutcome(ActivityExecutionOutcome.newBuilder().setResult(payloads("result", propagated))) + .build(); + } + + private static Payloads payloads(String value, PropagatedSerializationContext propagated) { + SerializationContext encodedWith = + propagated != null + ? NEXUS_CONTEXT + : new ActivitySerializationContext(NAMESPACE, null, null, ACTIVITY_TYPE, "tq", false); + return Payloads.newBuilder() + .addPayloads( + stamp( + DefaultDataConverter.STANDARD_INSTANCE + .toPayload(value) + .orElseThrow(() -> new IllegalStateException("expected payload")), + encodedWith)) + .build(); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/RootNexusClientInvokerSerializationContextTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/RootNexusClientInvokerSerializationContextTest.java new file mode 100644 index 0000000000..75b9dd30e4 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/RootNexusClientInvokerSerializationContextTest.java @@ -0,0 +1,136 @@ +package io.temporal.internal.client; + +import static io.temporal.internal.nexus.ContextRecordingCodec.stamp; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import io.grpc.Deadline; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.enums.v1.NexusOperationWaitStage; +import io.temporal.api.nexus.v1.PropagatedSerializationContext; +import io.temporal.api.workflowservice.v1.PollNexusOperationExecutionRequest; +import io.temporal.api.workflowservice.v1.PollNexusOperationExecutionResponse; +import io.temporal.client.NexusClientOptions; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.interceptors.NexusClientCallsInterceptor.GetNexusOperationResultInput; +import io.temporal.internal.client.external.GenericWorkflowClient; +import io.temporal.internal.nexus.ContextRecordingCodec; +import io.temporal.payload.context.NexusSerializationContext; +import java.util.Collections; +import java.util.concurrent.TimeUnit; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +/** + * Unit tests that a standalone Nexus operation's result is decoded with the context the server + * returns, falling back to the start request's context for older servers. + */ +public class RootNexusClientInvokerSerializationContextTest { + + private static final NexusSerializationContext SERVER_CONTEXT = + new NexusSerializationContext("server-endpoint", "server-service", "server-operation"); + private static final PropagatedSerializationContext PROPAGATED = + PropagatedSerializationContext.newBuilder() + .setEndpoint("server-endpoint") + .setService("server-service") + .setOperation("server-operation") + .build(); + private static final NexusSerializationContext START_CONTEXT = + new NexusSerializationContext("start-endpoint", "start-service", "start-operation"); + + private ContextRecordingCodec codec; + private GenericWorkflowClient genericClient; + private RootNexusClientInvoker invoker; + + @Before + public void setUp() { + codec = new ContextRecordingCodec(); + genericClient = mock(GenericWorkflowClient.class); + invoker = + new RootNexusClientInvoker( + genericClient, + new NexusClientResolvedOptions( + NexusClientOptions.getDefaultInstance().getNamespace(), + NexusClientOptions.getDefaultInstance().getInterceptors(), + codec.dataConverter(), + NexusClientOptions.getDefaultInstance().getIdentity())); + } + + @Test + public void serverContextTakesPrecedenceOverStartContext() throws Exception { + stubClosedPoll(SERVER_CONTEXT, PROPAGATED); + + String result = invoker.getNexusOperationResult(startedHandleInput()).getResult(); + + Assert.assertEquals("result", result); + Assert.assertEquals( + Collections.singletonList(ContextRecordingCodec.describe(SERVER_CONTEXT)), + codec.decodeContexts()); + } + + @Test + public void handleByIdUsesServerContext() throws Exception { + stubClosedPoll(SERVER_CONTEXT, PROPAGATED); + + invoker.getNexusOperationResult(handleByIdInput()); + + Assert.assertEquals( + Collections.singletonList(ContextRecordingCodec.describe(SERVER_CONTEXT)), + codec.decodeContexts()); + } + + @Test + public void olderServerFallsBackToStartContext() throws Exception { + stubClosedPoll(START_CONTEXT, null); + + invoker.getNexusOperationResult(startedHandleInput()); + + Assert.assertEquals( + Collections.singletonList(ContextRecordingCodec.describe(START_CONTEXT)), + codec.decodeContexts()); + } + + private void stubClosedPoll( + NexusSerializationContext encodedWith, PropagatedSerializationContext propagated) { + when(genericClient.pollNexusOperationExecution( + any(PollNexusOperationExecutionRequest.class), any(Deadline.class))) + .thenReturn(closedResponse(encodedWith, propagated)); + } + + private static PollNexusOperationExecutionResponse closedResponse( + NexusSerializationContext encodedWith, PropagatedSerializationContext propagated) { + Payload payload = + stamp( + DefaultDataConverter.STANDARD_INSTANCE + .toPayload("result") + .orElseThrow(() -> new IllegalStateException("expected payload")), + encodedWith); + PollNexusOperationExecutionResponse.Builder response = + PollNexusOperationExecutionResponse.newBuilder() + .setWaitStage(NexusOperationWaitStage.NEXUS_OPERATION_WAIT_STAGE_CLOSED) + .setResult(payload); + if (propagated != null) { + response.setPropagatedNexusSerializationContext(propagated); + } + return response.build(); + } + + private static GetNexusOperationResultInput startedHandleInput() { + return new GetNexusOperationResultInput<>( + "op-1", + null, + Deadline.after(10, TimeUnit.SECONDS), + String.class, + String.class, + START_CONTEXT.getEndpoint(), + START_CONTEXT.getService(), + START_CONTEXT.getOperation()); + } + + private static GetNexusOperationResultInput handleByIdInput() { + return new GetNexusOperationResultInput<>( + "op-1", null, Deadline.after(10, TimeUnit.SECONDS), String.class, String.class); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerNexusSerializationContextTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerNexusSerializationContextTest.java new file mode 100644 index 0000000000..6896e7dbf5 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerNexusSerializationContextTest.java @@ -0,0 +1,305 @@ +package io.temporal.internal.client; + +import static io.temporal.internal.nexus.ContextRecordingCodec.contextOf; +import static io.temporal.internal.nexus.ContextRecordingCodec.stamp; +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 com.uber.m3.tally.NoopScope; +import io.grpc.Deadline; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.Payloads; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.enums.v1.EventType; +import io.temporal.api.enums.v1.QueryRejectCondition; +import io.temporal.api.failure.v1.ApplicationFailureInfo; +import io.temporal.api.failure.v1.Failure; +import io.temporal.api.history.v1.History; +import io.temporal.api.history.v1.HistoryEvent; +import io.temporal.api.history.v1.WorkflowExecutionCompletedEventAttributes; +import io.temporal.api.history.v1.WorkflowExecutionContinuedAsNewEventAttributes; +import io.temporal.api.history.v1.WorkflowExecutionFailedEventAttributes; +import io.temporal.api.nexus.v1.PropagatedSerializationContext; +import io.temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryRequest; +import io.temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse; +import io.temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest; +import io.temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse; +import io.temporal.api.workflowservice.v1.StartWorkflowExecutionRequest; +import io.temporal.api.workflowservice.v1.StartWorkflowExecutionResponse; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowFailedException; +import io.temporal.client.WorkflowOptions; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.interceptors.Header; +import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.GetResultInput; +import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowSignalWithStartInput; +import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowStartInput; +import io.temporal.internal.client.external.GenericWorkflowClient; +import io.temporal.internal.nexus.ContextRecordingCodec; +import io.temporal.internal.nexus.CurrentNexusOperationContext; +import io.temporal.internal.nexus.InternalNexusOperationContext; +import io.temporal.payload.context.NexusSerializationContext; +import io.temporal.payload.context.WorkflowSerializationContext; +import java.util.Collections; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.mockito.ArgumentCaptor; + +/** + * Unit tests for propagating the Nexus serialization context to workflows started by a Nexus + * operation handler, and for decoding their results with the context the server returns. + */ +public class RootWorkflowClientInvokerNexusSerializationContextTest { + + private static final String NAMESPACE = "test-namespace"; + private static final String WORKFLOW_ID = "wf-target"; + private static final NexusSerializationContext NEXUS_CONTEXT = + new NexusSerializationContext("endpoint", "service", "operation"); + private static final PropagatedSerializationContext PROPAGATED = + PropagatedSerializationContext.newBuilder() + .setEndpoint("endpoint") + .setService("service") + .setOperation("operation") + .build(); + private static final String NEXUS = ContextRecordingCodec.describe(NEXUS_CONTEXT); + private static final String WORKFLOW = + ContextRecordingCodec.describe(new WorkflowSerializationContext(NAMESPACE, WORKFLOW_ID)); + + private ContextRecordingCodec codec; + private GenericWorkflowClient genericClient; + private RootWorkflowClientInvoker invoker; + + @Before + public void setUp() { + codec = new ContextRecordingCodec(); + genericClient = mock(GenericWorkflowClient.class); + invoker = + new RootWorkflowClientInvoker( + genericClient, + NAMESPACE, + "test-identity", + QueryRejectCondition.QUERY_REJECT_CONDITION_NONE, + Collections.emptyList(), + codec.dataConverter(), + new WorkerFactoryRegistry()); + } + + @After + public void tearDown() { + CurrentNexusOperationContext.unset(); + } + + @Test + public void startFromNexusHandlerPropagatesContextButKeepsWorkflowContextForPayloads() { + enterNexusHandler(NEXUS_CONTEXT); + when(genericClient.start(any())) + .thenReturn(StartWorkflowExecutionResponse.getDefaultInstance()); + + invoker.start(newStartInput()); + + StartWorkflowExecutionRequest sent = captureStartRequest(); + Assert.assertEquals(PROPAGATED, sent.getPropagatedNexusSerializationContext()); + // Only the workflow's result and failure, which are returned to the Nexus caller, use the + // propagated context. + Assert.assertEquals(WORKFLOW, contextOf(sent.getInput().getPayloads(0))); + Assert.assertEquals(WORKFLOW, contextOf(sent.getMemo().getFieldsOrThrow("memo-key"))); + Assert.assertEquals(WORKFLOW, contextOf(sent.getUserMetadata().getSummary())); + Assert.assertEquals(WORKFLOW, contextOf(sent.getUserMetadata().getDetails())); + } + + @Test + public void startOutsideNexusHandlerDoesNotPropagateContext() { + when(genericClient.start(any())) + .thenReturn(StartWorkflowExecutionResponse.getDefaultInstance()); + + invoker.start(newStartInput()); + + StartWorkflowExecutionRequest sent = captureStartRequest(); + Assert.assertFalse(sent.hasPropagatedNexusSerializationContext()); + Assert.assertEquals(WORKFLOW, contextOf(sent.getInput().getPayloads(0))); + } + + @Test + public void signalWithStartFromNexusHandlerKeepsWorkflowContext() { + enterNexusHandler(NEXUS_CONTEXT); + when(genericClient.signalWithStart(any())) + .thenReturn(SignalWithStartWorkflowExecutionResponse.getDefaultInstance()); + + invoker.signalWithStart( + new WorkflowSignalWithStartInput(newStartInput(), "signal", new Object[] {"signal-arg"})); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(SignalWithStartWorkflowExecutionRequest.class); + verify(genericClient).signalWithStart(captor.capture()); + Assert.assertEquals(WORKFLOW, contextOf(captor.getValue().getInput().getPayloads(0))); + Assert.assertEquals(WORKFLOW, contextOf(captor.getValue().getSignalInput().getPayloads(0))); + } + + @Test + public void getResultDecodesWithPropagatedContext() throws Exception { + when(genericClient.longPollHistory(any(GetWorkflowExecutionHistoryRequest.class), any())) + .thenReturn(historyResponse(completedEvent(NEXUS_CONTEXT), PROPAGATED)); + + String result = invoker.getResult(newGetResultInput()).getResult(); + + Assert.assertEquals("result", result); + Assert.assertEquals(Collections.singletonList(NEXUS), codec.decodeContexts()); + } + + @Test + public void getResultWithoutPropagatedContextUsesWorkflowContext() throws Exception { + when(genericClient.longPollHistory(any(GetWorkflowExecutionHistoryRequest.class), any())) + .thenReturn(historyResponse(completedEvent(null), null)); + + invoker.getResult(newGetResultInput()); + + Assert.assertEquals(Collections.singletonList(WORKFLOW), codec.decodeContexts()); + } + + @Test + public void getResultDecodesFailureWithPropagatedContext() { + when(genericClient.longPollHistory(any(GetWorkflowExecutionHistoryRequest.class), any())) + .thenReturn(historyResponse(failedEvent(NEXUS_CONTEXT), PROPAGATED)); + + Assert.assertThrows( + WorkflowFailedException.class, () -> invoker.getResult(newGetResultInput())); + Assert.assertEquals(Collections.singletonList(NEXUS), codec.decodeContexts()); + } + + @Test + public void getResultUsesContextOfTheRunThatClosed() throws Exception { + // The context is per run, so the one returned for a run that continued as new must not be + // applied to the next run. The server does carry the context into the continued run (see + // NexusSerializationContextPropagationTest); the next run here returns none so that a context + // leaked from the first run would be detected. + HistoryEvent continuedAsNew = + HistoryEvent.newBuilder() + .setEventType(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW) + .setWorkflowExecutionContinuedAsNewEventAttributes( + WorkflowExecutionContinuedAsNewEventAttributes.newBuilder() + .setNewExecutionRunId("run-2")) + .build(); + when(genericClient.longPollHistory(any(GetWorkflowExecutionHistoryRequest.class), any())) + .thenReturn(historyResponse(continuedAsNew, PROPAGATED)) + .thenReturn(historyResponse(completedEvent(null), null)); + + invoker.getResult(newGetResultInput()); + + Assert.assertEquals(Collections.singletonList(WORKFLOW), codec.decodeContexts()); + } + + @Test + public void getResultAsyncKeepsContextAcrossEmptyPolls() throws Exception { + // The context arrives on an empty long poll; the close event comes on the next one. + GetWorkflowExecutionHistoryResponse emptyPoll = + GetWorkflowExecutionHistoryResponse.newBuilder() + .setPropagatedNexusSerializationContext(PROPAGATED) + .build(); + when(genericClient.longPollHistoryAsync( + any(GetWorkflowExecutionHistoryRequest.class), any(Deadline.class))) + .thenReturn(CompletableFuture.completedFuture(emptyPoll)) + .thenReturn( + CompletableFuture.completedFuture( + historyResponse(completedEvent(NEXUS_CONTEXT), null))); + + invoker.getResultAsync(newGetResultInput()).getResult().get(); + + Assert.assertEquals(Collections.singletonList(NEXUS), codec.decodeContexts()); + } + + private void enterNexusHandler(NexusSerializationContext serializationContext) { + InternalNexusOperationContext nexusContext = + new InternalNexusOperationContext( + NAMESPACE, "tq", "endpoint", new NoopScope(), mock(WorkflowClient.class)); + if (serializationContext != null) { + nexusContext.setSerializationContext(serializationContext); + } + CurrentNexusOperationContext.set(nexusContext); + } + + private StartWorkflowExecutionRequest captureStartRequest() { + ArgumentCaptor captor = + ArgumentCaptor.forClass(StartWorkflowExecutionRequest.class); + verify(genericClient).start(captor.capture()); + return captor.getValue(); + } + + private static WorkflowStartInput newStartInput() { + WorkflowOptions options = + WorkflowOptions.newBuilder() + .setTaskQueue("tq") + .setDisableEagerExecution(true) + .setMemo(Collections.singletonMap("memo-key", "memo-value")) + .setStaticSummary("summary") + .setStaticDetails("details") + .build(); + return new WorkflowStartInput( + WORKFLOW_ID, "TestWorkflow", Header.empty(), new Object[] {"input"}, options); + } + + private static GetResultInput newGetResultInput() { + return new GetResultInput<>( + WorkflowExecution.newBuilder().setWorkflowId(WORKFLOW_ID).setRunId("run-1").build(), + Optional.of("TestWorkflow"), + 10, + TimeUnit.SECONDS, + String.class, + String.class); + } + + private static GetWorkflowExecutionHistoryResponse historyResponse( + HistoryEvent event, PropagatedSerializationContext propagated) { + GetWorkflowExecutionHistoryResponse.Builder response = + GetWorkflowExecutionHistoryResponse.newBuilder() + .setHistory(History.newBuilder().addEvents(event)); + if (propagated != null) { + response.setPropagatedNexusSerializationContext(propagated); + } + return response.build(); + } + + private static HistoryEvent completedEvent(NexusSerializationContext encodedWith) { + return HistoryEvent.newBuilder() + .setEventType(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED) + .setWorkflowExecutionCompletedEventAttributes( + WorkflowExecutionCompletedEventAttributes.newBuilder() + .setResult(payloads("result", encodedWith))) + .build(); + } + + private static HistoryEvent failedEvent(NexusSerializationContext encodedWith) { + Failure failure = + Failure.newBuilder() + .setMessage("failed") + .setApplicationFailureInfo( + ApplicationFailureInfo.newBuilder() + .setType("TestFailure") + .setDetails(payloads("detail", encodedWith))) + .build(); + return HistoryEvent.newBuilder() + .setEventType(EventType.EVENT_TYPE_WORKFLOW_EXECUTION_FAILED) + .setWorkflowExecutionFailedEventAttributes( + WorkflowExecutionFailedEventAttributes.newBuilder().setFailure(failure)) + .build(); + } + + private static Payloads payloads(String value, NexusSerializationContext encodedWith) { + Payload payload = + DefaultDataConverter.STANDARD_INSTANCE + .toPayload(value) + .orElseThrow(() -> new IllegalStateException("expected payload")); + return Payloads.newBuilder() + .addPayloads( + encodedWith == null + ? stamp(payload, new WorkflowSerializationContext(NAMESPACE, WORKFLOW_ID)) + : stamp(payload, encodedWith)) + .build(); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/nexus/ContextRecordingCodec.java b/temporal-sdk/src/test/java/io/temporal/internal/nexus/ContextRecordingCodec.java new file mode 100644 index 0000000000..05045aa353 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/nexus/ContextRecordingCodec.java @@ -0,0 +1,129 @@ +package io.temporal.internal.nexus; + +import com.google.protobuf.ByteString; +import io.temporal.api.common.v1.Payload; +import io.temporal.common.converter.CodecDataConverter; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.payload.codec.PayloadCodec; +import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.context.NexusSerializationContext; +import io.temporal.payload.context.SerializationContext; +import io.temporal.payload.context.WorkflowSerializationContext; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; + +/** + * Test codec that stamps the serialization context it encodes under onto each payload, and records + * the contexts it is asked to encode and decode under, so a test can tell which context was used + * where. + */ +public final class ContextRecordingCodec implements PayloadCodec { + public static final String CONTEXT_KEY = "test-serialization-context"; + + // Shared by every instance derived via withContext, so a test sees every encode and decode. + private final List encodeContexts; + private final List decodeContexts; + private final @Nullable SerializationContext context; + + public ContextRecordingCodec() { + this( + Collections.synchronizedList(new ArrayList<>()), + Collections.synchronizedList(new ArrayList<>()), + null); + } + + private ContextRecordingCodec( + List encodeContexts, + List decodeContexts, + @Nullable SerializationContext context) { + this.encodeContexts = encodeContexts; + this.decodeContexts = decodeContexts; + this.context = context; + } + + /** Forgets the contexts recorded so far. */ + public void reset() { + encodeContexts.clear(); + decodeContexts.clear(); + } + + /** A data converter that runs every payload through this codec. */ + public DataConverter dataConverter() { + return new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, Collections.singletonList(this)); + } + + /** The contexts payloads were encoded under, in the form returned by {@link #describe}. */ + public List encodeContexts() { + synchronized (encodeContexts) { + return new ArrayList<>(encodeContexts); + } + } + + /** The contexts payloads were decoded under, in the form returned by {@link #describe}. */ + public List decodeContexts() { + synchronized (decodeContexts) { + return new ArrayList<>(decodeContexts); + } + } + + /** The context the payload was encoded under, or null if it was not encoded by this codec. */ + public static @Nullable String contextOf(Payload payload) { + ByteString value = payload.getMetadataMap().get(CONTEXT_KEY); + return value == null ? null : value.toStringUtf8(); + } + + /** A short, comparable description of a serialization context. */ + public static String describe(@Nullable SerializationContext context) { + if (context instanceof NexusSerializationContext) { + NexusSerializationContext nexus = (NexusSerializationContext) context; + return "nexus:" + nexus.getEndpoint() + "/" + nexus.getService() + "/" + nexus.getOperation(); + } + if (context instanceof WorkflowSerializationContext) { + return "workflow:" + ((WorkflowSerializationContext) context).getWorkflowId(); + } + if (context instanceof ActivitySerializationContext) { + return "activity:" + ((ActivitySerializationContext) context).getActivityType(); + } + return "none"; + } + + /** Returns a copy of {@code payload} stamped as if encoded under {@code context}. */ + public static Payload stamp(Payload payload, @Nullable SerializationContext context) { + return Payload.newBuilder(payload) + .putMetadata(CONTEXT_KEY, ByteString.copyFromUtf8(describe(context))) + .build(); + } + + @Override + @Nonnull + public PayloadCodec withContext(@Nonnull SerializationContext context) { + return new ContextRecordingCodec(encodeContexts, decodeContexts, context); + } + + @Override + @Nonnull + public List encode(@Nonnull List payloads) { + encodeContexts.add(describe(context)); + List encoded = new ArrayList<>(payloads.size()); + for (Payload payload : payloads) { + encoded.add(stamp(payload, context)); + } + return encoded; + } + + @Override + @Nonnull + public List decode(@Nonnull List payloads) { + decodeContexts.add(describe(context)); + List decoded = new ArrayList<>(payloads.size()); + for (Payload payload : payloads) { + decoded.add(Payload.newBuilder(payload).removeMetadata(CONTEXT_KEY).build()); + } + return decoded; + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/sync/WorkflowNexusSerializationContextReplayTest.java b/temporal-sdk/src/test/java/io/temporal/internal/sync/WorkflowNexusSerializationContextReplayTest.java new file mode 100644 index 0000000000..cf1f38e68e --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/sync/WorkflowNexusSerializationContextReplayTest.java @@ -0,0 +1,250 @@ +package io.temporal.internal.sync; + +import io.temporal.api.history.v1.History; +import io.temporal.api.history.v1.HistoryEvent; +import io.temporal.api.nexus.v1.PropagatedSerializationContext; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.client.WorkflowFailedException; +import io.temporal.client.WorkflowOptions; +import io.temporal.client.WorkflowStub; +import io.temporal.client.WorkflowUpdateException; +import io.temporal.common.WorkflowExecutionHistory; +import io.temporal.failure.ApplicationFailure; +import io.temporal.internal.nexus.ContextRecordingCodec; +import io.temporal.payload.context.NexusSerializationContext; +import io.temporal.payload.context.WorkflowSerializationContext; +import io.temporal.testing.CloudTestExclusion.RequiresLocalServer; +import io.temporal.testing.CloudTestExclusionNote; +import io.temporal.testing.TestEnvironmentOptions; +import io.temporal.testing.TestWorkflowEnvironment; +import io.temporal.worker.Worker; +import io.temporal.workflow.UpdateMethod; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; +import java.util.Arrays; +import java.util.Collections; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +/** + * Tests that a workflow started by a Nexus operation handler encodes its result and closing + * failure, which are returned to the Nexus caller, with the Nexus context the server propagates in + * the WorkflowExecutionStarted event, and keeps the workflow context for everything else. + * + *

The test server doesn't propagate the context, so each test runs the workflow, adds the + * context to the started event of the recorded history, and replays it. + */ +@CloudTestExclusionNote("This test directly creates and controls a local test service.") +@Category(RequiresLocalServer.class) +public class WorkflowNexusSerializationContextReplayTest { + + private static final String TASK_QUEUE = "nexus-serialization-context-replay"; + private static final String NAMESPACE = "default"; + private static final NexusSerializationContext NEXUS_CONTEXT = + new NexusSerializationContext("endpoint", "service", "operation"); + private static final PropagatedSerializationContext PROPAGATED = + PropagatedSerializationContext.newBuilder() + .setEndpoint("endpoint") + .setService("service") + .setOperation("operation") + .build(); + private static final String NEXUS = ContextRecordingCodec.describe(NEXUS_CONTEXT); + + @WorkflowInterface + public interface EchoWorkflow { + @WorkflowMethod + String execute(String input); + + @UpdateMethod + void reject(); + } + + public static class EchoWorkflowImpl implements EchoWorkflow { + private boolean rejected; + + @Override + public String execute(String input) { + if ("update".equals(input)) { + Workflow.await(() -> rejected); + } + if ("fail".equals(input)) { + throw ApplicationFailure.newNonRetryableFailure("failed", "TestFailure", "detail"); + } + if ("memo".equals(input)) { + Workflow.upsertMemo(Collections.singletonMap("upserted", input)); + return (String) Workflow.getMemo("upserted", String.class); + } + if ("can".equals(input)) { + Workflow.continueAsNew("continued"); + } + return input; + } + + @Override + public void reject() { + rejected = true; + throw ApplicationFailure.newNonRetryableFailure("rejected", "TestFailure", "detail"); + } + } + + private ContextRecordingCodec codec; + private TestWorkflowEnvironment testEnvironment; + private Worker worker; + + @Before + public void setUp() { + codec = new ContextRecordingCodec(); + testEnvironment = + TestWorkflowEnvironment.newInstance( + TestEnvironmentOptions.newBuilder() + // With time skipping, the update in updateFailureKeepsWorkflowContext isn't + // delivered to the blocked workflow for several minutes. + .setUseTimeskipping(false) + .setWorkflowClientOptions( + WorkflowClientOptions.newBuilder() + .setNamespace(NAMESPACE) + .setDataConverter(codec.dataConverter()) + .build()) + .build()); + worker = testEnvironment.newWorker(TASK_QUEUE); + worker.registerWorkflowImplementationTypes(EchoWorkflowImpl.class); + testEnvironment.start(); + } + + @After + public void tearDown() { + testEnvironment.close(); + } + + @Test + public void propagatedContextIsUsedOnlyForResult() throws Exception { + WorkflowExecutionHistory history = run("hello"); + codec.reset(); + + worker.replayWorkflowExecution(withPropagatedContext(history)); + + // The input keeps the workflow context; the result is returned to the Nexus caller. + Assert.assertEquals( + Collections.singletonList(workflowContext("hello")), codec.decodeContexts()); + Assert.assertEquals(Collections.singletonList(NEXUS), codec.encodeContexts()); + } + + @Test + public void propagatedContextIsUsedForClosingFailure() throws Exception { + WorkflowExecutionHistory history = run("fail"); + codec.reset(); + + worker.replayWorkflowExecution(withPropagatedContext(history)); + + Assert.assertTrue( + "failure should be encoded with the Nexus context, saw " + codec.encodeContexts(), + codec.encodeContexts().contains(NEXUS)); + Assert.assertFalse(codec.encodeContexts().contains(workflowContext("fail"))); + } + + @Test + public void withoutPropagatedContextWorkflowContextIsUsed() throws Exception { + WorkflowExecutionHistory history = run("hello"); + codec.reset(); + + worker.replayWorkflowExecution(history); + + Assert.assertFalse(codec.decodeContexts().contains(NEXUS)); + Assert.assertFalse(codec.encodeContexts().contains(NEXUS)); + Assert.assertTrue(codec.encodeContexts().contains(workflowContext("hello"))); + } + + @Test + public void memoKeepsWorkflowContext() throws Exception { + WorkflowExecutionHistory history = run("memo"); + codec.reset(); + + worker.replayWorkflowExecution(withPropagatedContext(history)); + + // Upserting encodes the memo, and reading it back decodes it; only the result uses the Nexus + // context. + Assert.assertEquals(Arrays.asList(workflowContext("memo"), NEXUS), codec.encodeContexts()); + Assert.assertFalse(codec.decodeContexts().contains(NEXUS)); + } + + @Test + public void continueAsNewInputKeepsWorkflowContext() throws Exception { + WorkflowExecutionHistory history = run("can"); + codec.reset(); + + worker.replayWorkflowExecution(withPropagatedContext(history)); + + // The continued run decodes its input with the workflow context. + Assert.assertEquals(Collections.singletonList(workflowContext("can")), codec.encodeContexts()); + } + + @Test + public void updateFailureKeepsWorkflowContext() throws Exception { + WorkflowClient client = testEnvironment.getWorkflowClient(); + EchoWorkflow workflow = + client.newWorkflowStub( + EchoWorkflow.class, + WorkflowOptions.newBuilder() + .setTaskQueue(TASK_QUEUE) + .setWorkflowId(workflowId("update")) + .build()); + WorkflowClient.start(workflow::execute, "update"); + Assert.assertThrows(WorkflowUpdateException.class, workflow::reject); + WorkflowStub.fromTyped(workflow).getResult(String.class); + WorkflowExecutionHistory history = + client.fetchHistory( + workflowId("update"), WorkflowStub.fromTyped(workflow).getExecution().getRunId()); + codec.reset(); + + worker.replayWorkflowExecution(withPropagatedContext(history)); + + // The update failure belongs to the update, so it keeps the workflow context even though the + // workflow's result uses the Nexus context. + Assert.assertEquals(Arrays.asList(workflowContext("update"), NEXUS), codec.encodeContexts()); + } + + /** Runs the workflow to completion and returns the history of its first run. */ + private WorkflowExecutionHistory run(String input) { + WorkflowClient client = testEnvironment.getWorkflowClient(); + String workflowId = workflowId(input); + EchoWorkflow workflow = + client.newWorkflowStub( + EchoWorkflow.class, + WorkflowOptions.newBuilder() + .setTaskQueue(TASK_QUEUE) + .setWorkflowId(workflowId) + .build()); + try { + workflow.execute(input); + } catch (WorkflowFailedException e) { + // Expected for the failing workflow. + } + return client.fetchHistory( + workflowId, WorkflowStub.fromTyped(workflow).getExecution().getRunId()); + } + + /** Returns the history with the Nexus context added to its WorkflowExecutionStarted event. */ + private static WorkflowExecutionHistory withPropagatedContext(WorkflowExecutionHistory history) { + History.Builder builder = History.newBuilder().addAllEvents(history.getEvents()); + HistoryEvent.Builder started = builder.getEventsBuilder(0); + started + .getWorkflowExecutionStartedEventAttributesBuilder() + .setPropagatedNexusSerializationContext(PROPAGATED); + return new WorkflowExecutionHistory( + builder.build(), history.getWorkflowExecution().getWorkflowId()); + } + + private static String workflowId(String input) { + return "nexus-serialization-context-" + input; + } + + private static String workflowContext(String input) { + return ContextRecordingCodec.describe( + new WorkflowSerializationContext(NAMESPACE, workflowId(input))); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowSlotMaxConcurrentTests.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowSlotMaxConcurrentTests.java index 944d5d515e..f456a52622 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowSlotMaxConcurrentTests.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowSlotMaxConcurrentTests.java @@ -44,6 +44,12 @@ public class WorkflowSlotMaxConcurrentTests { SDKTestWorkflowRule.newBuilder() .setWorkerOptions( WorkerOptions.newBuilder() + // Explicit poller counts opt out of poller autoscaling auto-enrollment, which + // namespaces with the PollerAutoscalingAutoEnroll capability apply to pollers + // left + // at their defaults. + .setMaxConcurrentWorkflowTaskPollers(5) + .setMaxConcurrentActivityTaskPollers(5) .setMaxConcurrentWorkflowTaskExecutionSize( MAX_CONCURRENT_WORKFLOW_TASK_EXECUTION_SIZE) .setMaxConcurrentActivityExecutionSize(MAX_CONCURRENT_ACTIVITY_EXECUTION_SIZE) diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/nexus/NexusSerializationContextPropagationTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/NexusSerializationContextPropagationTest.java new file mode 100644 index 0000000000..51e18a7f86 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/NexusSerializationContextPropagationTest.java @@ -0,0 +1,601 @@ +package io.temporal.workflow.nexus; + +import static io.temporal.internal.nexus.ContextRecordingCodec.contextOf; + +import io.grpc.Deadline; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import io.nexusrpc.Operation; +import io.nexusrpc.Service; +import io.nexusrpc.handler.OperationHandler; +import io.nexusrpc.handler.OperationImpl; +import io.nexusrpc.handler.ServiceImpl; +import io.temporal.activity.Activity; +import io.temporal.activity.ActivityExecutionContext; +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import io.temporal.api.failure.v1.Failure; +import io.temporal.api.history.v1.HistoryEvent; +import io.temporal.api.history.v1.WorkflowExecutionStartedEventAttributes; +import io.temporal.api.nexus.v1.PropagatedSerializationContext; +import io.temporal.api.workflowservice.v1.DescribeActivityExecutionResponse; +import io.temporal.client.ActivityClient; +import io.temporal.client.ActivityClientOptions; +import io.temporal.client.DescribeActivityOptions; +import io.temporal.client.StartActivityOptions; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.client.WorkflowExecutionDescription; +import io.temporal.client.WorkflowOptions; +import io.temporal.common.RetryOptions; +import io.temporal.common.WorkflowExecutionHistory; +import io.temporal.failure.ApplicationFailure; +import io.temporal.failure.NexusOperationFailure; +import io.temporal.failure.TimeoutFailure; +import io.temporal.internal.nexus.ContextRecordingCodec; +import io.temporal.nexus.Nexus; +import io.temporal.nexus.TemporalOperationHandler; +import io.temporal.nexus.WorkflowRunOperation; +import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.context.NexusSerializationContext; +import io.temporal.payload.context.WorkflowSerializationContext; +import io.temporal.testing.CloudTestExclusion.RequiresCloudProvisioning; +import io.temporal.testing.CloudTestExclusionNote; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.workflow.NexusOperationOptions; +import io.temporal.workflow.NexusServiceOptions; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; +import java.time.Duration; +import java.util.Collections; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.junit.Assert; +import org.junit.Assume; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +/** + * End-to-end coverage that workflows and standalone activities started by a Nexus operation handler + * serialize their payloads with the handler's Nexus context, and that clients reading their results + * use the context the server returns. + * + *

Requires a server that propagates the context (temporalio/temporal#12400, #12401, #12403), + * with {@code nexusoperation.enableStandalone} and {@code activity.enableCallbacks} enabled. The + * dev server pinned in {@code SdkJavaTestServerProfile} includes them. + */ +@CloudTestExclusionNote("Requires a server with Nexus serialization context propagation.") +@Category(RequiresCloudProvisioning.class) +public class NexusSerializationContextPropagationTest { + + private static final ContextRecordingCodec CODEC = new ContextRecordingCodec(); + // Unique per run so a leftover execution from an earlier run can't be picked up. + private static final String RUN_ID = UUID.randomUUID().toString(); + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(CallerWorkflowImpl.class, BackingWorkflowImpl.class) + .setActivityImplementations( + new BackingActivityImpl(), new AsyncActivityImpl(), new StallingActivityImpl()) + .setNexusServiceImplementation(new ContextServiceImpl()) + .setWorkflowClientOptions( + WorkflowClientOptions.newBuilder().setDataConverter(CODEC.dataConverter()).build()) + .build(); + + @Before + public void setUp() { + Assume.assumeTrue( + "Nexus operations require a real server", SDKTestWorkflowRule.useExternalService); + CODEC.reset(); + } + + @Test + public void workflowStartedByHandlerUsesNexusContextForResult() { + String input = "workflow-" + RUN_ID; + Assert.assertEquals("hello " + input, newCaller().execute("startWorkflow", input)); + + String expected = describe("startWorkflow"); + String workflowContext = workflowContext(backingId(input)); + WorkflowExecutionHistory history = testWorkflowRule.getExecutionHistory(backingId(input)); + HistoryEvent started = history.getEvents().get(0); + Assert.assertEquals( + propagated("startWorkflow"), + started + .getWorkflowExecutionStartedEventAttributes() + .getPropagatedNexusSerializationContext()); + // Only the result, which is returned to the Nexus caller, uses the Nexus context. + Assert.assertEquals( + workflowContext, + contextOf(started.getWorkflowExecutionStartedEventAttributes().getInput().getPayloads(0))); + Assert.assertEquals( + expected, + contextOf( + history + .getLastEvent() + .getWorkflowExecutionCompletedEventAttributes() + .getResult() + .getPayloads(0))); + + // A detached handle gets the context from the server with the history. + CODEC.reset(); + String result = + testWorkflowRule + .getWorkflowClient() + .newUntypedWorkflowStub(backingId(input)) + .getResult(String.class); + Assert.assertEquals("hello " + input, result); + Assert.assertTrue( + "detached handle should decode with the Nexus context, saw " + CODEC.decodeContexts(), + CODEC.decodeContexts().contains(expected)); + + // Memo from the start request and the workflow's upsert, and user metadata, keep the workflow + // context, and describe decodes them with it. + Assert.assertEquals( + workflowContext, + contextOf( + started + .getWorkflowExecutionStartedEventAttributes() + .getMemo() + .getFieldsOrThrow("started"))); + CODEC.reset(); + WorkflowExecutionDescription description = + testWorkflowRule.getWorkflowClient().newUntypedWorkflowStub(backingId(input)).describe(); + Assert.assertEquals(input, description.getMemo("started", String.class)); + Assert.assertEquals(input, description.getMemo("upserted", String.class)); + Assert.assertEquals("summary-" + input, description.getStaticSummary()); + Assert.assertEquals("details-" + input, description.getStaticDetails()); + for (String context : CODEC.decodeContexts()) { + Assert.assertEquals( + "describe should decode with the workflow context", workflowContext, context); + } + } + + @Test + public void continuedAsNewRunKeepsNexusContext() { + String input = "can-" + RUN_ID; + Assert.assertEquals("hello " + input, newCaller().execute("startWorkflow", input)); + + String expected = describe("startWorkflow"); + // Without a run ID, the history is of the latest run, the one the first run continued into. + WorkflowExecutionHistory history = testWorkflowRule.getExecutionHistory(backingId(input)); + WorkflowExecutionStartedEventAttributes started = + history.getEvents().get(0).getWorkflowExecutionStartedEventAttributes(); + Assert.assertFalse( + "expected the latest run to be a continued run", + started.getContinuedExecutionRunId().isEmpty()); + Assert.assertEquals( + propagated("startWorkflow"), started.getPropagatedNexusSerializationContext()); + // Continue-as-new input keeps the workflow context; only the final result uses the Nexus one. + Assert.assertEquals( + workflowContext(backingId(input)), contextOf(started.getInput().getPayloads(0))); + Assert.assertEquals( + expected, + contextOf( + history + .getLastEvent() + .getWorkflowExecutionCompletedEventAttributes() + .getResult() + .getPayloads(0))); + + // A detached handle follows the continue-as-new chain and decodes the final result with the + // context the server returns for the run that closed. + CODEC.reset(); + String result = + testWorkflowRule + .getWorkflowClient() + .newUntypedWorkflowStub(backingId(input)) + .getResult(String.class); + Assert.assertEquals("hello " + input, result); + Assert.assertTrue( + "detached handle should decode with the Nexus context, saw " + CODEC.decodeContexts(), + CODEC.decodeContexts().contains(expected)); + } + + @Test + public void activityStartedByHandlerUsesNexusContextForResult() throws Exception { + String input = "activity-" + RUN_ID; + String expected = describe("startActivity"); + String activityContext = + ContextRecordingCodec.describe( + new ActivitySerializationContext( + testWorkflowRule.getWorkflowClient().getOptions().getNamespace(), + null, + null, + "Process", + testWorkflowRule.getTaskQueue(), + false)); + ActivityClient activityClient = newActivityClient(); + + BackingActivityImpl.release = new CountDownLatch(1); + CallerWorkflow caller = newCaller(); + CompletableFuture callerResult = + WorkflowClient.execute(caller::execute, "startActivity", input); + + // The heartbeat details and the first attempt's failure are read while the second attempt + // waits to be released, before the activity completes. + DescribeActivityExecutionResponse running = null; + Deadline deadline = Deadline.after(10, TimeUnit.SECONDS); + while (!deadline.isExpired()) { + try { + running = describeActivity(activityClient, input); + if (running.getInfo().hasLastFailure()) { + break; + } + } catch (StatusRuntimeException e) { + // The handler may not have started the activity yet. + if (e.getStatus().getCode() != Status.Code.NOT_FOUND) { + throw e; + } + } + Thread.sleep(100); + } + Assert.assertNotNull("activity was never described", running); + Assert.assertEquals( + propagated("startActivity"), running.getInfo().getPropagatedNexusSerializationContext()); + Assert.assertEquals(activityContext, contextOf(running.getInput().getPayloads(0))); + Assert.assertEquals( + activityContext, contextOf(running.getInfo().getUserMetadata().getSummary())); + // The server puts the heartbeat details into a timeout failure returned to the Nexus caller. + Assert.assertEquals( + expected, contextOf(running.getInfo().getHeartbeatDetails().getPayloads(0))); + // The attempt's failure, like the result, is returned to the Nexus caller. + Assert.assertEquals( + expected, + contextOf( + running + .getInfo() + .getLastFailure() + .getApplicationFailureInfo() + .getDetails() + .getPayloads(0))); + + BackingActivityImpl.release.countDown(); + Assert.assertEquals("hello " + input, callerResult.get(30, TimeUnit.SECONDS)); + Assert.assertEquals( + expected, + contextOf(describeActivity(activityClient, input).getOutcome().getResult().getPayloads(0))); + + // A handle obtained by ID gets the context from the server with the poll response. + CODEC.reset(); + Assert.assertEquals( + "hello " + input, + activityClient.getHandle(backingId(input), null, String.class).getResult()); + Assert.assertTrue( + "activity handle should decode with the Nexus context, saw " + CODEC.decodeContexts(), + CODEC.decodeContexts().contains(expected)); + } + + @Test + public void heartbeatTimeoutDetailsUseNexusContext() throws Exception { + String input = "stall-" + RUN_ID; + StallingActivityImpl.release = new CountDownLatch(1); + try { + CallerWorkflow caller = newCaller(); + String callerResult = caller.execute("startStallingActivity", input); + + // The caller gets the heartbeat timeout with the last heartbeat details. + Assert.assertEquals("heartbeat timeout: heartbeat-" + input, callerResult); + // The server copied the details, encoded with the Nexus context, into the timeout failure. + Failure failure = describeActivity(newActivityClient(), input).getOutcome().getFailure(); + Assert.assertEquals( + describe("startStallingActivity"), + contextOf(failure.getTimeoutFailureInfo().getLastHeartbeatDetails().getPayloads(0))); + } finally { + StallingActivityImpl.release.countDown(); + } + } + + @Test + public void asyncCompletedActivityUsesForwardedNexusContext() throws Exception { + String input = "async-" + RUN_ID; + AsyncActivityImpl.pending = new CompletableFuture<>(); + CallerWorkflow caller = newCaller(); + CompletableFuture callerResult = + WorkflowClient.execute(caller::execute, "startAsyncActivity", input); + + // The activity hands its task token and Nexus context to the test, as it would to an + // external system, and the test completes it with that context. + PendingActivity pending = AsyncActivityImpl.pending.get(30, TimeUnit.SECONDS); + Assert.assertEquals(nexusContext("startAsyncActivity"), pending.nexusContext); + testWorkflowRule + .getWorkflowClient() + .newActivityCompletionClient() + .withContext(pending.nexusContext) + .complete(pending.taskToken, "hello " + input); + + Assert.assertEquals("hello " + input, callerResult.get(30, TimeUnit.SECONDS)); + Assert.assertEquals( + describe("startAsyncActivity"), + contextOf( + describeActivity(newActivityClient(), input).getOutcome().getResult().getPayloads(0))); + } + + private ActivityClient newActivityClient() { + return ActivityClient.newInstance( + testWorkflowRule.getWorkflowServiceStubs(), + ActivityClientOptions.newBuilder() + .setNamespace(testWorkflowRule.getWorkflowClient().getOptions().getNamespace()) + .setDataConverter(CODEC.dataConverter()) + .build()); + } + + private static DescribeActivityExecutionResponse describeActivity( + ActivityClient activityClient, String input) { + return activityClient + .getHandle(backingId(input), null) + .describe( + DescribeActivityOptions.newBuilder() + .setIncludeInput(true) + .setIncludeOutcome(true) + .setIncludeHeartbeatDetails(true) + .setIncludeLastFailure(true) + .build()) + .getRawResponse(); + } + + private CallerWorkflow newCaller() { + return testWorkflowRule.newWorkflowStubTimeoutOptions(CallerWorkflow.class); + } + + private String workflowContext(String workflowId) { + return ContextRecordingCodec.describe( + new WorkflowSerializationContext( + testWorkflowRule.getWorkflowClient().getOptions().getNamespace(), workflowId)); + } + + private String describe(String operation) { + return ContextRecordingCodec.describe(nexusContext(operation)); + } + + private NexusSerializationContext nexusContext(String operation) { + return new NexusSerializationContext( + testWorkflowRule.getNexusEndpoint().getSpec().getName(), + ContextService.class.getSimpleName(), + operation); + } + + private PropagatedSerializationContext propagated(String operation) { + return PropagatedSerializationContext.newBuilder() + .setEndpoint(testWorkflowRule.getNexusEndpoint().getSpec().getName()) + .setService(ContextService.class.getSimpleName()) + .setOperation(operation) + .build(); + } + + /** ID of the workflow or activity backing the operation called with {@code input}. */ + private static String backingId(String input) { + return "backing-" + input; + } + + @Service + public interface ContextService { + @Operation + String startWorkflow(String input); + + @Operation + String startActivity(String input); + + @Operation + String startAsyncActivity(String input); + + @Operation + String startStallingActivity(String input); + } + + @ServiceImpl(service = ContextService.class) + public static class ContextServiceImpl { + @OperationImpl + public OperationHandler startWorkflow() { + return WorkflowRunOperation.fromWorkflowMethod( + (context, details, input) -> + Nexus.getOperationContext() + .getWorkflowClient() + .newWorkflowStub( + BackingWorkflow.class, + WorkflowOptions.newBuilder() + .setWorkflowId(backingId(input)) + .setMemo(Collections.singletonMap("started", input)) + .setStaticSummary("summary-" + input) + .setStaticDetails("details-" + input) + .setTaskQueue(Nexus.getOperationContext().getInfo().getTaskQueue()) + .build()) + ::execute); + } + + @OperationImpl + public OperationHandler startActivity() { + return TemporalOperationHandler.create( + (context, client, input) -> + client.startActivity( + BackingActivity.class, + BackingActivity::process, + input, + StartActivityOptions.newBuilder() + .setId(backingId(input)) + .setSummary("summary-" + input) + .setRetryOptions( + RetryOptions.newBuilder() + .setInitialInterval(Duration.ofMillis(100)) + .build()) + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build())); + } + + @OperationImpl + public OperationHandler startAsyncActivity() { + return TemporalOperationHandler.create( + (context, client, input) -> + client.startActivity( + AsyncActivity.class, + AsyncActivity::process, + input, + StartActivityOptions.newBuilder() + .setId(backingId(input)) + .setStartToCloseTimeout(Duration.ofSeconds(30)) + .build())); + } + + @OperationImpl + public OperationHandler startStallingActivity() { + return TemporalOperationHandler.create( + (context, client, input) -> + client.startActivity( + StallingActivity.class, + StallingActivity::process, + input, + StartActivityOptions.newBuilder() + .setId(backingId(input)) + .setHeartbeatTimeout(Duration.ofSeconds(1)) + .setStartToCloseTimeout(Duration.ofSeconds(30)) + .setRetryOptions(RetryOptions.newBuilder().setMaximumAttempts(1).build()) + .build())); + } + } + + @WorkflowInterface + public interface CallerWorkflow { + @WorkflowMethod + String execute(String operation, String input); + } + + public static class CallerWorkflowImpl implements CallerWorkflow { + @Override + public String execute(String operation, String input) { + ContextService service = + Workflow.newNexusServiceStub( + ContextService.class, + NexusServiceOptions.newBuilder() + .setOperationOptions( + NexusOperationOptions.newBuilder() + .setScheduleToCloseTimeout(Duration.ofSeconds(30)) + .build()) + .build()); + try { + switch (operation) { + case "startWorkflow": + return service.startWorkflow(input); + case "startAsyncActivity": + return service.startAsyncActivity(input); + case "startStallingActivity": + return service.startStallingActivity(input); + default: + return service.startActivity(input); + } + } catch (NexusOperationFailure e) { + for (Throwable cause = e; cause != null; cause = cause.getCause()) { + if (cause instanceof TimeoutFailure) { + return "heartbeat timeout: " + + ((TimeoutFailure) cause).getLastHeartbeatDetails().get(String.class); + } + } + return "failed: " + e.getCause().getMessage(); + } + } + } + + @WorkflowInterface + public interface BackingWorkflow { + @WorkflowMethod + String execute(String input); + } + + public static class BackingWorkflowImpl implements BackingWorkflow { + @Override + public String execute(String input) { + Workflow.upsertMemo(Collections.singletonMap("upserted", input)); + // Continue as new once, so the result comes from the second run. + if (input.startsWith("can-") + && !Workflow.getInfo().getContinuedExecutionRunId().isPresent()) { + Workflow.continueAsNew(input); + } + return "hello " + input; + } + } + + @ActivityInterface + public interface BackingActivity { + @ActivityMethod + String process(String input); + } + + public static class BackingActivityImpl implements BackingActivity { + // Holds the second attempt until the test has described the running activity. + static volatile CountDownLatch release = new CountDownLatch(0); + + @Override + public String process(String input) { + // The first attempt fails after heartbeating, so the worker sends the heartbeat details with + // the failure, and the server keeps them along with the attempt's failure. + if (Activity.getExecutionContext().getInfo().getAttempt() == 1) { + Activity.getExecutionContext().heartbeat("heartbeat-" + input); + throw ApplicationFailure.newFailure("first attempt fails", "Retry", "detail-" + input); + } + try { + release.await(30, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return "hello " + input; + } + } + + @ActivityInterface + public interface AsyncActivity { + @ActivityMethod(name = "ProcessAsync") + String process(String input); + } + + /** The task token and Nexus context an activity hands off to be completed asynchronously. */ + static final class PendingActivity { + final byte[] taskToken; + final NexusSerializationContext nexusContext; + + PendingActivity(byte[] taskToken, NexusSerializationContext nexusContext) { + this.taskToken = taskToken; + this.nexusContext = nexusContext; + } + } + + public static class AsyncActivityImpl implements AsyncActivity { + static volatile CompletableFuture pending = new CompletableFuture<>(); + + @Override + public String process(String input) { + ActivityExecutionContext context = Activity.getExecutionContext(); + pending.complete( + new PendingActivity( + context.getInfo().getTaskToken(), context.getInfo().getNexusSerializationContext())); + context.doNotCompleteOnReturn(); + return null; + } + } + + @ActivityInterface + public interface StallingActivity { + @ActivityMethod(name = "ProcessStalling") + String process(String input); + } + + public static class StallingActivityImpl implements StallingActivity { + // Holds the activity, after its one heartbeat, until the test is done. + static volatile CountDownLatch release = new CountDownLatch(0); + + @Override + public String process(String input) { + // Heartbeat once, then stop heartbeating, so the attempt times out with these details. + Activity.getExecutionContext().heartbeat("heartbeat-" + input); + try { + release.await(30, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return "hello " + input; + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/nexus/OperationFailMetricTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/OperationFailMetricTest.java index 4fd89f0e73..55badb8097 100644 --- a/temporal-sdk/src/test/java/io/temporal/workflow/nexus/OperationFailMetricTest.java +++ b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/OperationFailMetricTest.java @@ -172,6 +172,10 @@ public void failOperationApplicationErrorMetrics() { applicationFailure.getOriginalMessage()); Assert.assertEquals("OperationError", applicationFailure.getType()); Assert.assertNotNull(applicationFailure.getCause()); + } + // Newer servers convert an operation error in the old format to the new format, wrapping the + // handler's failure in an OperationError. + if ("OperationError".equals(applicationFailure.getType())) { applicationFailure = (ApplicationFailure) applicationFailure.getCause(); } Assert.assertEquals("intentional failure", applicationFailure.getOriginalMessage()); @@ -252,6 +256,10 @@ public void failOperationMessageApplicationErrorMetrics() { if (isUsingNewFormat()) { Assert.assertEquals("failure message", applicationFailure.getOriginalMessage()); Assert.assertEquals("OperationError", applicationFailure.getType()); + } + // Newer servers convert an operation error in the old format to the new format, wrapping the + // handler's failure in an OperationError. + if ("OperationError".equals(applicationFailure.getType())) { applicationFailure = (ApplicationFailure) applicationFailure.getCause(); } Assert.assertEquals("intentional failure", applicationFailure.getOriginalMessage()); diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/queryTests/LocalActivityAndQueryTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/queryTests/LocalActivityAndQueryTest.java index b8be13df1c..ccad0cbd77 100644 --- a/temporal-sdk/src/test/java/io/temporal/workflow/queryTests/LocalActivityAndQueryTest.java +++ b/temporal-sdk/src/test/java/io/temporal/workflow/queryTests/LocalActivityAndQueryTest.java @@ -12,6 +12,7 @@ import io.temporal.common.converter.DefaultDataConverter; import io.temporal.testing.internal.SDKTestOptions; import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.worker.WorkerOptions; import io.temporal.workflow.Workflow; import io.temporal.workflow.shared.TestActivities.TestActivitiesImpl; import io.temporal.workflow.shared.TestActivities.VariousTestActivities; @@ -34,6 +35,14 @@ public class LocalActivityAndQueryTest { public SDKTestWorkflowRule testWorkflowRule = SDKTestWorkflowRule.newBuilder() .setTestTimeoutSeconds(30) + // Explicit poller counts opt out of poller autoscaling auto-enrollment, which namespaces + // with the PollerAutoscalingAutoEnroll capability apply to pollers left at their + // defaults. Queries sent during the long workflow task then don't expire. + .setWorkerOptions( + WorkerOptions.newBuilder() + .setMaxConcurrentWorkflowTaskPollers(5) + .setMaxConcurrentActivityTaskPollers(5) + .build()) .setWorkflowTypes(TestLocalActivityAndQueryWorkflow.class) .setActivityImplementations(activitiesImpl) .build(); diff --git a/temporal-testing/src/main/java/io/temporal/testing/internal/devserver/SdkJavaTestServerProfile.java b/temporal-testing/src/main/java/io/temporal/testing/internal/devserver/SdkJavaTestServerProfile.java index 334eb42550..37073eb4bf 100644 --- a/temporal-testing/src/main/java/io/temporal/testing/internal/devserver/SdkJavaTestServerProfile.java +++ b/temporal-testing/src/main/java/io/temporal/testing/internal/devserver/SdkJavaTestServerProfile.java @@ -13,7 +13,7 @@ public final class SdkJavaTestServerProfile { public static final String ACTIVE_PROPERTY = "io.temporal.testing.internal.devServerProfile"; // This is intentionally the sole Temporal CLI version used by sdk-java repository tests. - private static final String TEST_CLI_VERSION = "1.8.3-server-1.32.0-162.0"; + private static final String TEST_CLI_VERSION = "1.9.3-nexus-serialization-context"; private static final String TEST_NAMESPACE = "UnitTest"; private static final String DATABASE_FILENAME = "temporal.sqlite"; @@ -117,6 +117,10 @@ private static List repositoryServerArguments() { "frontend.ListWorkersEnabled=true", "--dynamic-config-value", "frontend.enableCancelWorkerPollsOnShutdown=true", + // Matching fan-out poll cancellation can drop a query dispatched while a stopped worker's + // polls are canceled. + "--dynamic-config-value", + "frontend.enableMatchingFanOutForPollCancellation=false", "--dynamic-config-value", "component.callbacks.allowedAddresses=[{\"Pattern\":\"localhost:7243\",\"AllowInsecure\":true}]", "--dynamic-config-value",