diff --git a/temporal-sdk/src/main/java/io/temporal/client/NexusOperationExecutionDescription.java b/temporal-sdk/src/main/java/io/temporal/client/NexusOperationExecutionDescription.java index 50fd5837ba..2e9ddfc337 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/NexusOperationExecutionDescription.java +++ b/temporal-sdk/src/main/java/io/temporal/client/NexusOperationExecutionDescription.java @@ -24,12 +24,24 @@ public final class NexusOperationExecutionDescription extends NexusOperationExec private final DescribeNexusOperationExecutionResponse response; private final NexusOperationExecutionInfo info; - private final DataConverter dataConverter; + private final DataConverter dataConverterWithNexusContext; + // User metadata is attached by the caller without a Nexus serialization context, so it has to be + // decoded without one too. Everything else on a description belongs to the operation and is + // decoded with the operation's context. + private final DataConverter contextlessDataConverter; public NexusOperationExecutionDescription( DescribeNexusOperationExecutionResponse response, DataConverter dataConverter, String namespace) { + this(response, dataConverter, dataConverter, namespace); + } + + public NexusOperationExecutionDescription( + DescribeNexusOperationExecutionResponse response, + DataConverter dataConverterWithNexusContext, + DataConverter contextlessDataConverter, + String namespace) { super( null, response.getInfo().getOperationId(), @@ -51,7 +63,8 @@ public NexusOperationExecutionDescription( : null); this.response = response; this.info = response.getInfo(); - this.dataConverter = dataConverter; + this.dataConverterWithNexusContext = dataConverterWithNexusContext; + this.contextlessDataConverter = contextlessDataConverter; } /** Underlying proto response. Exposed while the Nexus SDK surface is still experimental. */ @@ -124,7 +137,7 @@ public Instant getLastAttemptCompleteTime() { @Nullable public Exception getLastAttemptFailure() { return info.hasLastAttemptFailure() - ? dataConverter.failureToException(info.getLastAttemptFailure()) + ? dataConverterWithNexusContext.failureToException(info.getLastAttemptFailure()) : null; } @@ -140,7 +153,8 @@ public Instant getNextAttemptScheduleTime() { @Nullable public NexusOperationCancellationInfo getCancellationInfo() { return info.hasCancellationInfo() - ? new NexusOperationCancellationInfo(info.getCancellationInfo(), dataConverter) + ? new NexusOperationCancellationInfo( + info.getCancellationInfo(), dataConverterWithNexusContext) : null; } @@ -183,7 +197,7 @@ public String getStaticSummary() { if (!info.hasUserMetadata() || !info.getUserMetadata().hasSummary()) { return null; } - return dataConverter.fromPayload( + return contextlessDataConverter.fromPayload( info.getUserMetadata().getSummary(), String.class, String.class); } @@ -196,7 +210,7 @@ public String getStaticDetails() { if (!info.hasUserMetadata() || !info.getUserMetadata().hasDetails()) { return null; } - return dataConverter.fromPayload( + return contextlessDataConverter.fromPayload( info.getUserMetadata().getDetails(), String.class, String.class); } @@ -231,7 +245,7 @@ public Optional getInput(Class valueType, Type genericType) { return Optional.empty(); } return Optional.ofNullable( - dataConverter.fromPayload(response.getInput(), valueType, genericType)); + dataConverterWithNexusContext.fromPayload(response.getInput(), valueType, genericType)); } /** @@ -266,7 +280,7 @@ public Optional getResult(Class valueType, Type genericType) { return Optional.empty(); } return Optional.ofNullable( - dataConverter.fromPayload(response.getResult(), valueType, genericType)); + dataConverterWithNexusContext.fromPayload(response.getResult(), valueType, genericType)); } /** @@ -275,6 +289,8 @@ public Optional getResult(Class valueType, Type genericType) { */ @Nullable public Exception getFailure() { - return response.hasFailure() ? dataConverter.failureToException(response.getFailure()) : null; + return response.hasFailure() + ? dataConverterWithNexusContext.failureToException(response.getFailure()) + : null; } } diff --git a/temporal-sdk/src/main/java/io/temporal/client/UntypedNexusServiceClientImpl.java b/temporal-sdk/src/main/java/io/temporal/client/UntypedNexusServiceClientImpl.java index b7dd9334a9..73222e3ade 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/UntypedNexusServiceClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/client/UntypedNexusServiceClientImpl.java @@ -8,6 +8,7 @@ import io.temporal.common.interceptors.NexusClientCallsInterceptor.StartNexusOperationExecutionOutput; import io.temporal.internal.client.NexusClientResolvedOptions; import io.temporal.internal.client.NexusOperationHandleImpl; +import io.temporal.payload.context.NexusSerializationContext; import java.lang.reflect.Type; import java.util.Collections; import javax.annotation.Nullable; @@ -43,12 +44,15 @@ class UntypedNexusServiceClientImpl implements UntypedNexusServiceClient { @Override public UntypedNexusOperationHandle start( String operation, StartNexusOperationOptions options, @Nullable Object arg) { - Payload payload = serializeInput(arg); + Payload payload = serializeInput(arg, operation); StartNexusOperationExecutionInput input = new StartNexusOperationExecutionInput( endpoint, serviceName, operation, payload, options, Collections.emptyMap()); StartNexusOperationExecutionOutput output = invoker.startNexusOperationExecution(input); - return new NexusOperationHandleImpl(output.getOperationId(), output.getRunId(), invoker); + // The handle keeps what the start request was for, including when the server returned an + // operation that was already running, so the result is decoded the way it was encoded. + return new NexusOperationHandleImpl( + output.getOperationId(), output.getRunId(), invoker, endpoint, serviceName, operation); } @Override @@ -71,12 +75,13 @@ public R execute( return NexusOperationHandle.fromUntyped(handle, resultClass, resultType).getResult(); } - private @Nullable Payload serializeInput(@Nullable Object arg) { + private @Nullable Payload serializeInput(@Nullable Object arg, String operation) { if (arg == null) { return null; } Class argClass = arg.getClass(); return dataConverter + .withContext(new NexusSerializationContext(endpoint, serviceName, operation)) .toPayload(arg) .orElseThrow( () -> diff --git a/temporal-sdk/src/main/java/io/temporal/common/interceptors/NexusClientCallsInterceptor.java b/temporal-sdk/src/main/java/io/temporal/common/interceptors/NexusClientCallsInterceptor.java index 9af27865bc..5f0bd59980 100644 --- a/temporal-sdk/src/main/java/io/temporal/common/interceptors/NexusClientCallsInterceptor.java +++ b/temporal-sdk/src/main/java/io/temporal/common/interceptors/NexusClientCallsInterceptor.java @@ -10,6 +10,7 @@ import io.temporal.client.NexusOperationHandle; import io.temporal.client.StartNexusOperationOptions; import io.temporal.common.Experimental; +import io.temporal.payload.context.NexusSerializationContext; import java.lang.reflect.Type; import java.util.Collections; import java.util.Map; @@ -250,18 +251,65 @@ final class GetNexusOperationResultInput { private final @Nonnull Deadline deadline; private final Class resultClass; private final @Nullable Type resultType; + private final @Nullable String endpoint; + private final @Nullable String service; + private final @Nullable String operation; + /** + * Equivalent to {@link #GetNexusOperationResultInput(String, String, Deadline, Class, Type, + * String, String, String)} with no endpoint, service or operation, which is the case for a + * handle obtained by operation ID rather than by starting an operation. + */ public GetNexusOperationResultInput( String operationId, @Nullable String runId, @Nonnull Deadline deadline, Class resultClass, @Nullable Type resultType) { + this(operationId, runId, deadline, resultClass, resultType, null, null, null); + } + + /** + * The endpoint, service and operation identify the Nexus operation the result is being read + * for, and are used to build the {@link NexusSerializationContext} the result and failure are + * decoded with. They must all be set or all be {@code null}: a partially identified operation + * would silently decode without a context, which for a converter that varies by context means + * reading the payload the wrong way rather than failing. + * + * @param endpoint Nexus endpoint the operation was started on, or {@code null} if the operation + * was not started through this handle + * @param service Nexus service the operation was started on, or {@code null} + * @param operation Nexus operation that was started, or {@code null} + * @throws IllegalArgumentException if only some of endpoint, service and operation are set + */ + public GetNexusOperationResultInput( + String operationId, + @Nullable String runId, + @Nonnull Deadline deadline, + Class resultClass, + @Nullable Type resultType, + @Nullable String endpoint, + @Nullable String service, + @Nullable String operation) { + boolean anySet = endpoint != null || service != null || operation != null; + boolean allSet = endpoint != null && service != null && operation != null; + if (anySet && !allSet) { + throw new IllegalArgumentException( + "endpoint, service and operation must all be set or all be null, got endpoint=" + + endpoint + + ", service=" + + service + + ", operation=" + + operation); + } this.operationId = operationId; this.runId = runId; this.deadline = deadline; this.resultClass = resultClass; this.resultType = resultType; + this.endpoint = endpoint; + this.service = service; + this.operation = operation; } public String getOperationId() { @@ -285,6 +333,34 @@ public Class getResultClass() { public Type getResultType() { return resultType; } + + /** + * Nexus endpoint the operation was started on. {@code null} when the operation was not started + * through this handle, in which case {@link #getService()} and {@link #getOperation()} are + * {@code null} too and the result is decoded without a Nexus serialization context. + */ + @Nullable + public String getEndpoint() { + return endpoint; + } + + /** + * Nexus service the operation was started on, or {@code null}. Set exactly when {@link + * #getEndpoint()} is set. + */ + @Nullable + public String getService() { + return service; + } + + /** + * Nexus operation that was started, or {@code null}. Set exactly when {@link #getEndpoint()} is + * set. + */ + @Nullable + public String getOperation() { + return operation; + } } final class GetNexusOperationResultOutput { diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/NexusOperationHandleImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/client/NexusOperationHandleImpl.java index 732de7e49c..ac6ca98f07 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/NexusOperationHandleImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/NexusOperationHandleImpl.java @@ -25,9 +25,25 @@ public final class NexusOperationHandleImpl implements UntypedNexusOperationHand private final String operationId; private final @Nullable String runId; private final NexusClientCallsInterceptor interceptor; + // What the operation was started on, retained so its result and failure are decoded with the + // same serialization context that encoded them. All null for a handle obtained by operation ID, + // which never saw a start request. + private final @Nullable String endpoint; + private final @Nullable String service; + private final @Nullable String operation; public NexusOperationHandleImpl( String operationId, @Nullable String runId, NexusClientCallsInterceptor interceptor) { + this(operationId, runId, interceptor, null, null, null); + } + + public NexusOperationHandleImpl( + String operationId, + @Nullable String runId, + NexusClientCallsInterceptor interceptor, + @Nullable String endpoint, + @Nullable String service, + @Nullable String operation) { if (operationId == null) { throw new IllegalArgumentException("operationId is required"); } @@ -37,6 +53,9 @@ public NexusOperationHandleImpl( this.operationId = operationId; this.runId = runId; this.interceptor = interceptor; + this.endpoint = endpoint; + this.service = service; + this.operation = operation; } @Override @@ -116,7 +135,14 @@ public R getResult( throws TimeoutException { GetNexusOperationResultInput input = new GetNexusOperationResultInput<>( - operationId, runId, Deadline.after(timeout, unit), resultClass, resultType); + operationId, + runId, + Deadline.after(timeout, unit), + resultClass, + resultType, + endpoint, + service, + operation); return interceptor.getNexusOperationResult(input).getResult(); } @@ -131,7 +157,14 @@ public CompletableFuture getResultAsync( long timeout, TimeUnit unit, Class resultClass, @Nullable Type resultType) { GetNexusOperationResultInput input = new GetNexusOperationResultInput<>( - operationId, runId, Deadline.after(timeout, unit), resultClass, resultType); + operationId, + runId, + Deadline.after(timeout, unit), + resultClass, + resultType, + endpoint, + service, + operation); return interceptor .getNexusOperationResultAsync(input) .thenApply(GetNexusOperationResultOutput::getResult); 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 415dbceea1..8d519b21db 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 @@ -8,6 +8,7 @@ import io.temporal.api.enums.v1.NexusOperationWaitStage; import io.temporal.api.errordetails.v1.NexusOperationExecutionAlreadyStartedFailure; import io.temporal.api.failure.v1.Failure; +import io.temporal.api.nexus.v1.NexusOperationExecutionInfo; import io.temporal.api.sdk.v1.UserMetadata; import io.temporal.api.workflowservice.v1.CountNexusOperationExecutionsRequest; import io.temporal.api.workflowservice.v1.CountNexusOperationExecutionsResponse; @@ -28,10 +29,12 @@ import io.temporal.client.NexusOperationNotFoundException; import io.temporal.client.StartNexusOperationOptions; import io.temporal.common.Experimental; +import io.temporal.common.converter.DataConverter; import io.temporal.common.interceptors.NexusClientCallsInterceptor; import io.temporal.internal.client.external.GenericWorkflowClient; import io.temporal.internal.common.ProtobufTimeUtils; import io.temporal.internal.common.WorkflowExecutionUtils; +import io.temporal.payload.context.NexusSerializationContext; import io.temporal.serviceclient.StatusUtils; import java.util.Iterator; import java.util.Objects; @@ -140,9 +143,42 @@ public DescribeNexusOperationExecutionOutput describeNexusOperationExecution( } catch (StatusRuntimeException e) { throw mapNotFound(input.getOperationId(), input.getRunId().orElse(null), e); } + // The response names the endpoint, service and operation, so the description decodes its + // payloads and failures with the same context the operation was started with. + NexusOperationExecutionInfo info = response.getInfo(); + DataConverter dataConverter = + clientOptions + .getDataConverter() + .withContext( + new NexusSerializationContext( + info.getEndpoint(), info.getService(), info.getOperation())); return new DescribeNexusOperationExecutionOutput( new NexusOperationExecutionDescription( - response, clientOptions.getDataConverter(), clientOptions.getNamespace())); + response, + dataConverter, + // The summary and details were attached without a Nexus context, so a converter that + // varies by context only round-trips them if they are decoded without one too. + clientOptions.getDataConverter(), + clientOptions.getNamespace())); + } + + /** + * 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. + * + *

{@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) { + DataConverter dataConverter = clientOptions.getDataConverter(); + if (input.getEndpoint() == null) { + return dataConverter; + } + return dataConverter.withContext( + new NexusSerializationContext( + input.getEndpoint(), input.getService(), input.getOperation())); } private DescribeNexusOperationExecutionRequest buildDescribeRequest( @@ -246,13 +282,14 @@ private GetNexusOperationResultOutput extractResult( @Nullable String runId, PollNexusOperationExecutionResponse response, GetNexusOperationResultInput input) { + DataConverter dataConverter = dataConverterFor(input); if (response.hasFailure()) { Failure failure = response.getFailure(); throw new NexusOperationFailedException( "Nexus operation failed: operationId='" + operationId + "'", operationId, runId, - clientOptions.getDataConverter().failureToException(failure)); + dataConverter.failureToException(failure)); } if (!response.hasResult()) { throw new NexusOperationFailedException( @@ -266,12 +303,10 @@ private GetNexusOperationResultOutput extractResult( } Payload payload = response.getResult(); R deserialized = - clientOptions - .getDataConverter() - .fromPayload( - payload, - input.getResultClass(), - input.getResultType() != null ? input.getResultType() : input.getResultClass()); + dataConverter.fromPayload( + payload, + input.getResultClass(), + input.getResultType() != null ? input.getResultType() : input.getResultClass()); return new GetNexusOperationResultOutput<>(deserialized); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java b/temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java index 97ab4f5ed2..0fa7002013 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java @@ -6,10 +6,12 @@ import io.temporal.common.interceptors.NexusOperationOutboundCallsInterceptor; import io.temporal.nexus.NexusOperationContext; import io.temporal.nexus.NexusOperationInfo; +import io.temporal.payload.context.NexusSerializationContext; import java.util.ArrayList; import java.util.Collections; import java.util.List; import javax.annotation.Nonnull; +import javax.annotation.Nullable; public class InternalNexusOperationContext { private final String namespace; @@ -39,6 +41,10 @@ public class InternalNexusOperationContext { private final List responseLinks = new ArrayList<>(); private NexusOperationMetadata nexusOperationMetadata; + // Serialization context for the operation this task is for. Set by the task handler once the + // service and operation names are known, which is only after the request variant has been + // inspected, so it is null while the task is being dispatched. + private NexusSerializationContext serializationContext; public InternalNexusOperationContext( String namespace, @@ -93,6 +99,22 @@ public NexusOperationMetadata getNexusOperationMetadata() { return nexusOperationMetadata; } + /** + * Sets the serialization context describing the operation this task is for. Called by the task + * handler once the request variant has been inspected and the service and operation are known. + */ + public void setSerializationContext(NexusSerializationContext serializationContext) { + this.serializationContext = serializationContext; + } + + /** + * Serialization context for the operation this task is for, or {@code null} if the service and + * operation are not known yet. + */ + public @Nullable NexusSerializationContext getSerializationContext() { + return serializationContext; + } + /** * Set the {@code common.v1.Link}s extracted from the inbound Nexus task so they can be attached * to RPCs issued by the operation handler. diff --git a/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusTaskHandlerImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusTaskHandlerImpl.java index 4d40183c27..620681ea0a 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusTaskHandlerImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusTaskHandlerImpl.java @@ -26,6 +26,7 @@ import io.temporal.internal.worker.NexusTask; import io.temporal.internal.worker.NexusTaskHandler; import io.temporal.internal.worker.ShutdownManager; +import io.temporal.payload.context.NexusSerializationContext; import io.temporal.serviceclient.CheckedExceptionWrapper; import io.temporal.worker.TypeAlreadyRegisteredException; import java.net.URISyntaxException; @@ -154,6 +155,27 @@ public Result handle(NexusTask task, Scope metricsScope) throws TimeoutException } } + /** + * Records the serialization context for the operation this task is for, so that the data + * converter used for its input, result and failures is scoped to the endpoint, service and + * operation the request names. + */ + private void setSerializationContext(String service, String operation) { + InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.get(); + nexusContext.setSerializationContext( + new NexusSerializationContext(nexusContext.getEndpoint(), service, operation)); + } + + /** + * The data converter scoped to the operation this task is for. Falls back to the uncontextualized + * converter if the request variant did not name a service and operation. + */ + private DataConverter dataConverterForCurrentOperation() { + NexusSerializationContext context = + CurrentNexusOperationContext.get().getSerializationContext(); + return context != null ? dataConverter.withContext(context) : dataConverter; + } + private void cancelOperation(OperationContext context, OperationCancelDetails details) { try { serviceHandler.cancelOperation(context, details); @@ -173,6 +195,7 @@ private void cancelOperation(OperationContext context, OperationCancelDetails de private CancelOperationResponse handleCancelledOperation( OperationContext.Builder ctx, CancelOperationRequest task) { ctx.setService(task.getService()).setOperation(task.getOperation()); + setSerializationContext(task.getService(), task.getOperation()); @SuppressWarnings("deprecation") // getOperationId kept to support old server for a while OperationCancelDetails operationCancelDetails = @@ -281,6 +304,7 @@ private OperationStartResult startOperation( private StartOperationResponse handleStartOperation( OperationContext.Builder ctx, StartOperationRequest task) { ctx.setService(task.getService()).setOperation(task.getOperation()); + setSerializationContext(task.getService(), task.getOperation()); OperationStartDetails.Builder operationStartDetails = OperationStartDetails.newBuilder() @@ -379,7 +403,8 @@ private StartOperationResponse handleStartOperation( HandlerException.ErrorType.INTERNAL, new RuntimeException("Unknown operation state: " + e.getState())); } - startResponseBuilder.setFailure(dataConverter.exceptionToFailure(temporalFailure)); + startResponseBuilder.setFailure( + dataConverterForCurrentOperation().exceptionToFailure(temporalFailure)); } return startResponseBuilder.build(); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/nexus/PayloadSerializer.java b/temporal-sdk/src/main/java/io/temporal/internal/nexus/PayloadSerializer.java index 97127768cb..ab6429a213 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/nexus/PayloadSerializer.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/nexus/PayloadSerializer.java @@ -7,6 +7,7 @@ import io.temporal.common.converter.DataConverter; import io.temporal.common.converter.DataConverterException; import io.temporal.failure.ApplicationFailure; +import io.temporal.payload.context.NexusSerializationContext; import java.lang.reflect.ParameterizedType; import java.lang.reflect.Type; import java.util.Optional; @@ -47,9 +48,26 @@ class PayloadSerializer implements Serializer { this.dataConverter = dataConverter; } + /** + * The data converter scoped to the operation currently being handled. + * + *

A single serializer is shared by every operation the worker handles, so the context is + * resolved per call from the task being handled rather than captured once. Falls back to the + * uncontextualized converter when there is no Nexus task in scope, which is the case when this + * serializer is used directly rather than by the task handler. + */ + private DataConverter dataConverter() { + if (!CurrentNexusOperationContext.isNexusContext()) { + return dataConverter; + } + NexusSerializationContext context = + CurrentNexusOperationContext.get().getSerializationContext(); + return context != null ? dataConverter.withContext(context) : dataConverter; + } + @Override public Content serialize(@Nullable Object o) { - Optional payload = dataConverter.toPayload(o); + Optional payload = dataConverter().toPayload(o); Content.Builder content = Content.newBuilder(); content.setData(payload.get().toByteArray()); return content.build(); @@ -59,6 +77,7 @@ public Content serialize(@Nullable Object o) { public @Nullable Object deserialize(Content content, Type type) { try { Payload payload = Payload.parseFrom(content.getData()); + DataConverter dataConverter = dataConverter(); if ((type instanceof Class)) { return dataConverter.fromPayload(payload, (Class) type, type); } else if (type instanceof ParameterizedType) { 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 065ce71428..e38c6ea4d3 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 @@ -46,6 +46,7 @@ import io.temporal.internal.replay.WorkflowContext; import io.temporal.internal.statemachines.*; import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.context.NexusSerializationContext; import io.temporal.payload.context.WorkflowSerializationContext; import io.temporal.worker.WorkflowImplementationOptions; import io.temporal.workflow.*; @@ -798,9 +799,16 @@ public ExecuteNexusOperationOutput executeNexusOperation( CompletablePromise operationPromise = Workflow.newPromise(); CompletablePromise> resultPromise = Workflow.newPromise(); - // Not using the context aware data converter because the context will not be available on the - // worker side - Optional payload = dataConverter.toPayload(input.getArg()); + // The caller workflow is not available to the operation handler, so Nexus payloads are + // contextualized by the endpoint, service and operation instead. The same converter decodes the + // result and converts failures, so each operation keeps the converter selected for it even when + // several operations are in flight at once. + DataConverter nexusDataConverter = + dataConverter.withContext( + new NexusSerializationContext( + input.getEndpoint(), input.getService(), input.getOperation())); + + Optional payload = nexusDataConverter.toPayload(input.getArg()); ScheduleNexusOperationCommandAttributes.Builder attributes = ScheduleNexusOperationCommandAttributes.newBuilder(); @@ -835,7 +843,7 @@ public ExecuteNexusOperationOutput executeNexusOperation( "nexus operation start failed callback", () -> operationPromise.completeExceptionally( - dataConverter.failureToException(failure))); + nexusDataConverter.failureToException(failure))); } else { runner.executeInWorkflowThread( "nexus operation started callback", @@ -849,7 +857,7 @@ public ExecuteNexusOperationOutput executeNexusOperation( "nexus operation failure callback", () -> resultPromise.completeExceptionally( - dataConverter.failureToException(failure))); + nexusDataConverter.failureToException(failure))); } else { runner.executeInWorkflowThread( "nexus operation completion callback", () -> resultPromise.complete(result)); @@ -869,7 +877,7 @@ public ExecuteNexusOperationOutput executeNexusOperation( resultPromise.thenApply( (b) -> input.getResultClass() != Void.class - ? dataConverter.fromPayload( + ? nexusDataConverter.fromPayload( b.get(), input.getResultClass(), input.getResultType()) : null); // We register an empty handler to make sure that this promise is always "accessed" and never 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 new file mode 100644 index 0000000000..3e355c1a14 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/payload/context/NexusSerializationContext.java @@ -0,0 +1,104 @@ +package io.temporal.payload.context; + +import io.temporal.common.Experimental; +import io.temporal.common.converter.FailureConverter; +import java.util.Objects; +import javax.annotation.Nonnull; + +/** + * {@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. + * + *

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. + * + *

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. + */ +@Experimental +public final class NexusSerializationContext implements SerializationContext { + private final @Nonnull String endpoint; + private final @Nonnull String service; + private final @Nonnull String operation; + + /** + * @param endpoint the Nexus endpoint name; must not be {@code null} + * @param service the Nexus service name; must not be {@code null} + * @param operation the resolved Nexus operation name; must not be {@code null} + */ + public NexusSerializationContext( + @Nonnull String endpoint, @Nonnull String service, @Nonnull String operation) { + this.endpoint = Objects.requireNonNull(endpoint, "endpoint"); + this.service = Objects.requireNonNull(service, "service"); + this.operation = Objects.requireNonNull(operation, "operation"); + } + + /** + * @return the Nexus endpoint name + */ + @Nonnull + public String getEndpoint() { + return endpoint; + } + + /** + * @return the Nexus service name + */ + @Nonnull + public String getService() { + return service; + } + + /** + * @return the resolved Nexus operation name + */ + @Nonnull + public String getOperation() { + return operation; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof NexusSerializationContext)) { + return false; + } + NexusSerializationContext that = (NexusSerializationContext) o; + return endpoint.equals(that.endpoint) + && service.equals(that.service) + && operation.equals(that.operation); + } + + @Override + public int hashCode() { + return Objects.hash(endpoint, service, operation); + } + + @Override + public String toString() { + return "NexusSerializationContext{" + + "endpoint='" + + endpoint + + '\'' + + ", service='" + + service + + '\'' + + ", operation='" + + operation + + '\'' + + '}'; + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/payload/context/SerializationContext.java b/temporal-sdk/src/main/java/io/temporal/payload/context/SerializationContext.java index 0dfc254cc4..0b4d4f1a7c 100644 --- a/temporal-sdk/src/main/java/io/temporal/payload/context/SerializationContext.java +++ b/temporal-sdk/src/main/java/io/temporal/payload/context/SerializationContext.java @@ -36,8 +36,9 @@ * PayloadConverter#withContext(SerializationContext)} and using the modified instance when * applicable. * - *

Nexus operations inside a workflow do NOT have a {@link WorkflowSerializationContext} because - * it is not available in the operation handler. + *

Nexus operation payloads get a {@link NexusSerializationContext} rather than a {@link + * WorkflowSerializationContext}, because the caller workflow is not available in the operation + * handler. * *

Note: Serialization Context is experimental feature, the class and field structure of {@link * SerializationContext} objects may change in the future. There may be also situation where the 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 new file mode 100644 index 0000000000..8ee35c8999 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/client/nexus/StandaloneNexusSerializationContextTest.java @@ -0,0 +1,374 @@ +package io.temporal.client.nexus; + +import static org.junit.Assume.assumeTrue; + +import com.google.protobuf.ByteString; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.nexus.v1.Endpoint; +import io.temporal.client.NexusClient; +import io.temporal.client.NexusClientOptions; +import io.temporal.client.NexusOperationExecutionDescription; +import io.temporal.client.NexusOperationFailedException; +import io.temporal.client.StartNexusOperationOptions; +import io.temporal.client.UntypedNexusOperationHandle; +import io.temporal.client.UntypedNexusServiceClient; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.common.converter.CodecDataConverter; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.converter.FailureConverter; +import io.temporal.failure.DefaultFailureConverter; +import io.temporal.payload.codec.PayloadCodec; +import io.temporal.payload.context.NexusSerializationContext; +import io.temporal.payload.context.SerializationContext; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.workflow.shared.EchoNexusServiceImpl; +import io.temporal.workflow.shared.TestWorkflows; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.UUID; +import javax.annotation.Nonnull; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; + +/** + * Coverage that the standalone Nexus client scopes its data converter to the endpoint, service and + * operation of the operation it is acting on, and that a handle obtained by operation ID — which + * has no endpoint, service or operation — serializes without a context instead. + * + *

Standalone Nexus operations require a real server with them enabled. + */ +public class StandaloneNexusSerializationContextTest { + private static final String SERVICE = "TestNexusService1"; + private static final String OPERATION = "operation"; + + // Both sides share the codec so a signature written by one is checked by the other. Any payload + // the SDK encodes and decodes under different contexts therefore fails the decode, the way a + // codec keyed on the context would. Only the client gets the recording failure converter, so the + // contexts it records are the client's. + private static final RecordingCodec CODEC = new RecordingCodec(); + private static final RecordingFailureConverter FAILURE_CONVERTER = + new RecordingFailureConverter(); + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(PlaceholderWorkflowImpl.class) + .setNexusServiceImplementation(new EchoNexusServiceImpl()) + .setWorkflowClientOptions( + WorkflowClientOptions.newBuilder() + .setDataConverter( + new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, Collections.singletonList(CODEC))) + .build()) + .build(); + + private NexusClient nexusClient() { + return NexusClient.newInstance( + testWorkflowRule.getWorkflowServiceStubs(), + NexusClientOptions.newBuilder() + .setNamespace(testWorkflowRule.getWorkflowClient().getOptions().getNamespace()) + .setDataConverter( + new CodecDataConverter( + DefaultDataConverter.newDefaultInstance() + .withFailureConverter(FAILURE_CONVERTER), + Collections.singletonList(CODEC))) + .build()); + } + + @Before + public void requireStandaloneNexusSupport() { + assumeTrue( + "server does not support standalone Nexus operations", + testWorkflowRule.isUseExternalService()); + CODEC.reset(); + FAILURE_CONVERTER.reset(); + } + + @Test + public void startedHandleUsesItsStartRequestContext() { + String input = "ping-" + UUID.randomUUID(); + UntypedNexusOperationHandle handle = startOperation(input); + + // Decoding the result correctly is itself the assertion: the codec rejects a payload whose + // recorded context does not match the context it is being decoded under. + Assert.assertEquals("echo:" + input, handle.getResult(String.class)); + Assert.assertTrue( + "the start input and the polled result should both use the operation's context, but saw " + + CODEC.nexusContexts(), + CODEC.nexusContexts().contains(expectedContext())); + } + + @Test + public void failureUsesTheOperationsContext() { + UntypedNexusOperationHandle handle = + startOperation(EchoNexusServiceImpl.FAIL_PREFIX + UUID.randomUUID()); + // Ignore what the start request itself converted, so only the failure path is observed. + FAILURE_CONVERTER.reset(); + + Assert.assertThrows(NexusOperationFailedException.class, () -> handle.getResult(String.class)); + Assert.assertEquals( + "the operation failure should be converted under the operation's context", + Collections.singletonList(expectedContext()), + FAILURE_CONVERTER.nexusContexts()); + } + + @Test + public void describeUsesContextFromTheResponse() { + String input = "ping-" + UUID.randomUUID(); + UntypedNexusOperationHandle handle = startOperation(input); + handle.getResult(String.class); + CODEC.reset(); + + // A description decodes its payloads lazily, so reading one is what exercises the converter it + // was built with. + NexusOperationExecutionDescription description = handle.describe(); + Assert.assertEquals( + java.util.Optional.of("echo:" + input), description.getResult(String.class)); + + Assert.assertTrue( + "describe should build the context from the endpoint, service and operation the server " + + "reports, but saw " + + CODEC.allContexts(), + CODEC.nexusContexts().contains(expectedContext())); + } + + @Test + public void describeDecodesTheLastAttemptFailureWithContext() { + UntypedNexusOperationHandle handle = + startOperation(EchoNexusServiceImpl.FAIL_PREFIX + UUID.randomUUID()); + Assert.assertThrows(NexusOperationFailedException.class, () -> handle.getResult(String.class)); + FAILURE_CONVERTER.reset(); + + NexusOperationExecutionDescription description = handle.describe(); + Assert.assertNotNull("expected a terminal failure to describe", description.getFailure()); + + Assert.assertEquals( + "the described failure should be converted under the context the server reported", + Collections.singletonList(expectedContext()), + FAILURE_CONVERTER.nexusContexts()); + } + + @Test + public void describeReadsTheUncontextualizedSummary() { + NexusClient client = nexusClient(); + Endpoint endpoint = testWorkflowRule.getNexusEndpoint(); + UntypedNexusServiceClient serviceClient = + client.newUntypedNexusServiceClient(endpoint.getSpec().getName(), SERVICE); + UntypedNexusOperationHandle handle = + serviceClient.start( + OPERATION, + StartNexusOperationOptions.newBuilder() + .setId(UUID.randomUUID().toString()) + .setScheduleToCloseTimeout(Duration.ofSeconds(30)) + .setSummary("the-summary") + .build(), + "ping-" + UUID.randomUUID()); + handle.getResult(String.class); + + // The summary is encoded without a Nexus context, so describe must read it back the same way. + // Decoding it under a context the encoder never used would corrupt it. + Assert.assertEquals("the-summary", handle.describe().getStaticSummary()); + } + + @Test + public void handleObtainedByIdHasNoContext() { + 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. + 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(), + CODEC.nexusContexts()); + } + + private NexusSerializationContext expectedContext() { + return new NexusSerializationContext( + testWorkflowRule.getNexusEndpoint().getSpec().getName(), SERVICE, OPERATION); + } + + private UntypedNexusOperationHandle startOperation(String input) { + NexusClient client = nexusClient(); + Endpoint endpoint = testWorkflowRule.getNexusEndpoint(); + UntypedNexusServiceClient serviceClient = + client.newUntypedNexusServiceClient(endpoint.getSpec().getName(), SERVICE); + StartNexusOperationOptions options = + StartNexusOperationOptions.newBuilder() + .setId(UUID.randomUUID().toString()) + .setScheduleToCloseTimeout(Duration.ofSeconds(30)) + .build(); + return serviceClient.start(OPERATION, options, input); + } + + public static class PlaceholderWorkflowImpl implements TestWorkflows.TestWorkflow1 { + @Override + public String execute(String input) { + return input; + } + } + + /** Records the Nexus contexts the SDK scopes failure conversion by. */ + private static class RecordingFailureConverter implements FailureConverter { + private final List seen; + private final SerializationContext context; + private final FailureConverter delegate = new DefaultFailureConverter(); + + RecordingFailureConverter() { + this(Collections.synchronizedList(new ArrayList<>()), null); + } + + private RecordingFailureConverter( + List seen, SerializationContext context) { + this.seen = seen; + this.context = context; + } + + void reset() { + seen.clear(); + } + + List nexusContexts() { + List result = new ArrayList<>(); + synchronized (seen) { + for (SerializationContext each : seen) { + if (each instanceof NexusSerializationContext) { + result.add((NexusSerializationContext) each); + } + } + } + return result; + } + + @Override + @Nonnull + public FailureConverter withContext(@Nonnull SerializationContext context) { + return new RecordingFailureConverter(seen, context); + } + + @Override + @Nonnull + public RuntimeException failureToException( + @Nonnull io.temporal.api.failure.v1.Failure failure, @Nonnull DataConverter dataConverter) { + seen.add(context); + return delegate.failureToException(failure, dataConverter); + } + + @Override + @Nonnull + public io.temporal.api.failure.v1.Failure exceptionToFailure( + @Nonnull Throwable throwable, @Nonnull DataConverter dataConverter) { + seen.add(context); + return delegate.exceptionToFailure(throwable, dataConverter); + } + } + + /** + * Records the Nexus contexts it is handed, and tags each payload it encodes with the context + * used, refusing to decode a payload under a context other than the one that encoded it. + */ + private static class RecordingCodec implements PayloadCodec { + private static final String SIGNATURE_KEY = "ser-ctx-signature"; + + // Shared by every instance derived via withContext, so a test sees all contexts that were used. + private final List seen; + private final SerializationContext context; + + RecordingCodec() { + this(Collections.synchronizedList(new ArrayList<>()), null); + } + + private RecordingCodec(List seen, SerializationContext context) { + this.seen = seen; + this.context = context; + } + + void reset() { + seen.clear(); + } + + List allContexts() { + synchronized (seen) { + return new ArrayList<>(seen); + } + } + + List nexusContexts() { + List result = new ArrayList<>(); + synchronized (seen) { + for (SerializationContext each : seen) { + if (each instanceof NexusSerializationContext) { + result.add((NexusSerializationContext) each); + } + } + } + return result; + } + + @Override + @Nonnull + public PayloadCodec withContext(@Nonnull SerializationContext context) { + return new RecordingCodec(seen, context); + } + + @Override + @Nonnull + public List encode(@Nonnull List payloads) { + seen.add(context); + if (!(context instanceof NexusSerializationContext)) { + return payloads; + } + NexusSerializationContext nexus = (NexusSerializationContext) context; + String signature = + nexus.getEndpoint() + ":" + nexus.getService() + ":" + nexus.getOperation(); + List encoded = new ArrayList<>(payloads.size()); + for (Payload payload : payloads) { + encoded.add( + Payload.newBuilder(payload) + .putMetadata(SIGNATURE_KEY, ByteString.copyFromUtf8(signature)) + .build()); + } + return encoded; + } + + @Override + @Nonnull + public List decode(@Nonnull List payloads) { + seen.add(context); + List decoded = new ArrayList<>(payloads.size()); + for (Payload payload : payloads) { + ByteString signature = payload.getMetadataMap().get(SIGNATURE_KEY); + if (signature == null) { + // Decoding under a Nexus context something that was encoded without one means the SDK + // picked different contexts for the two halves of a round trip. A codec keyed on the + // context, such as a per-endpoint encryption key, could not recover this payload. + Assert.assertFalse( + "payload encoded without a context was decoded under " + context, + context instanceof NexusSerializationContext); + decoded.add(payload); + continue; + } + 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()); + } + decoded.add(Payload.newBuilder(payload).removeMetadata(SIGNATURE_KEY).build()); + } + return decoded; + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/RootNexusClientInvokerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/RootNexusClientInvokerTest.java index 6f3cea502c..4b4779a34f 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/client/RootNexusClientInvokerTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/RootNexusClientInvokerTest.java @@ -37,6 +37,50 @@ public class RootNexusClientInvokerTest { NexusClientOptions.getDefaultInstance().getDataConverter(), NexusClientOptions.getDefaultInstance().getIdentity())); + @Test + public void resultInputRejectsPartiallyIdentifiedOperation() { + // A partially identified operation would decode without a Nexus context, which for a converter + // that varies by context means reading the payload the wrong way rather than failing. + Assert.assertThrows( + IllegalArgumentException.class, + () -> + new GetNexusOperationResultInput<>( + "op-1", + null, + Deadline.after(10, TimeUnit.SECONDS), + String.class, + String.class, + "endpoint", + null, + "operation")); + } + + @Test + public void resultInputAcceptsFullyIdentifiedOperation() { + GetNexusOperationResultInput input = + new GetNexusOperationResultInput<>( + "op-1", + null, + Deadline.after(10, TimeUnit.SECONDS), + String.class, + String.class, + "endpoint", + "service", + "operation"); + Assert.assertEquals("endpoint", input.getEndpoint()); + Assert.assertEquals("service", input.getService()); + Assert.assertEquals("operation", input.getOperation()); + } + + @Test + public void resultInputAcceptsUnidentifiedOperation() { + // A handle obtained by operation ID never saw a start request. + GetNexusOperationResultInput input = input(); + Assert.assertNull(input.getEndpoint()); + Assert.assertNull(input.getService()); + Assert.assertNull(input.getOperation()); + } + private static GetNexusOperationResultInput input() { 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/nexus/NexusTaskHandlerSerializationContextTest.java b/temporal-sdk/src/test/java/io/temporal/internal/nexus/NexusTaskHandlerSerializationContextTest.java new file mode 100644 index 0000000000..bebb0cc3ac --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/nexus/NexusTaskHandlerSerializationContextTest.java @@ -0,0 +1,261 @@ +package io.temporal.internal.nexus; + +import static org.mockito.Mockito.mock; + +import com.google.protobuf.ByteString; +import com.uber.m3.tally.RootScopeBuilder; +import com.uber.m3.tally.Scope; +import com.uber.m3.util.Duration; +import io.nexusrpc.OperationException; +import io.nexusrpc.handler.OperationHandler; +import io.nexusrpc.handler.OperationImpl; +import io.nexusrpc.handler.ServiceImpl; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.failure.v1.Failure; +import io.temporal.api.nexus.v1.Request; +import io.temporal.api.nexus.v1.StartOperationRequest; +import io.temporal.api.workflowservice.v1.PollNexusTaskQueueResponse; +import io.temporal.client.WorkflowClient; +import io.temporal.common.converter.CodecDataConverter; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.interceptors.WorkerInterceptor; +import io.temporal.common.reporter.TestStatsReporter; +import io.temporal.internal.worker.NexusTask; +import io.temporal.internal.worker.NexusTaskHandler; +import io.temporal.payload.codec.PayloadCodec; +import io.temporal.payload.context.NexusSerializationContext; +import io.temporal.payload.context.SerializationContext; +import io.temporal.workflow.shared.TestNexusServices; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.TimeoutException; +import javax.annotation.Nonnull; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +/** + * Verifies that the Nexus task handler scopes its data converter to the endpoint, service and + * operation the inbound request names, for operation input, synchronous results and failures. + */ +public class NexusTaskHandlerSerializationContextTest { + private static final String NAMESPACE = "testNamespace"; + private static final String TASK_QUEUE = "testTaskQueue"; + private static final String ENDPOINT = "handler-endpoint"; + private static final String SERVICE = "TestNexusService1"; + private static final String OPERATION = "operation"; + + private Scope metricsScope; + + @Before + public void setUp() { + metricsScope = + new RootScopeBuilder().reporter(new TestStatsReporter()).reportEvery(Duration.ofMillis(10)); + } + + @Test + public void inputAndSyncResultUseOperationContext() throws TimeoutException { + NexusSerializationContext expected = + new NexusSerializationContext(ENDPOINT, SERVICE, OPERATION); + // A separate converter stands in for the caller, so the contexts recorded by the handler's own + // codec are only the ones the handler used. + DataConverter callerConverter = signingConverter(new SigningCodec()); + SigningCodec handlerCodec = new SigningCodec(); + DataConverter handlerConverter = signingConverter(handlerCodec); + + // The caller encodes the input under the operation's context, so the handler has to decode it + // under the same context to read it back. + Payload input = callerConverter.withContext(expected).toPayload("handler-input").get(); + + NexusTaskHandler.Result result = + handle(handlerConverter, new EchoServiceImpl(), startTask(input)); + + Assert.assertNull(result.getHandlerException()); + Payload resultPayload = result.getResponse().getStartOperation().getSyncSuccess().getPayload(); + Assert.assertEquals( + "the sync result should be encoded under the operation's context", + signature(expected), + resultPayload.getMetadataOrThrow(SigningCodec.SIGNATURE_KEY).toStringUtf8()); + Assert.assertEquals( + "Hello, handler-input!", + callerConverter + .withContext(expected) + .fromPayload(resultPayload, String.class, String.class)); + Assert.assertEquals( + "the handler should decode the input and encode the result under the operation's context", + java.util.Arrays.asList(expected, expected), + handlerCodec.contexts()); + } + + @Test + public void operationFailureUsesOperationContext() throws TimeoutException { + NexusSerializationContext expected = + new NexusSerializationContext(ENDPOINT, SERVICE, OPERATION); + DataConverter callerConverter = signingConverter(new SigningCodec()); + SigningCodec handlerCodec = new SigningCodec(); + DataConverter handlerConverter = signingConverter(handlerCodec); + Payload input = callerConverter.withContext(expected).toPayload("boom").get(); + + NexusTaskHandler.Result result = + handle(handlerConverter, new FailingServiceImpl(), startTask(input)); + + Assert.assertNull(result.getHandlerException()); + Failure failure = result.getResponse().getStartOperation().getFailure(); + Assert.assertNotEquals( + "the operation should have reported a failure", Failure.getDefaultInstance(), failure); + Assert.assertEquals( + "every converter call the handler made should be under the operation's context", + Collections.singleton(expected), + new java.util.HashSet<>(handlerCodec.contexts())); + } + + @Test + public void serializerWithoutTaskInScopeUsesContextlessConverter() { + // The serializer is shared by the whole worker and is also reachable outside of a Nexus task, + // where there is no endpoint/service/operation to scope it by. + SigningCodec codec = new SigningCodec(); + DataConverter dataConverter = signingConverter(codec); + + PayloadSerializer serializer = new PayloadSerializer(dataConverter); + serializer.serialize("no-task-in-scope"); + + Assert.assertEquals( + "no Nexus task is in scope, so the codec should be called without a context", + Collections.singletonList(null), + codec.contexts()); + } + + private static DataConverter signingConverter(SigningCodec codec) { + return new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, Collections.singletonList(codec)); + } + + private NexusTaskHandler.Result handle( + DataConverter dataConverter, Object serviceImpl, PollNexusTaskQueueResponse.Builder task) + throws TimeoutException { + NexusTaskHandlerImpl handler = + new NexusTaskHandlerImpl( + mock(WorkflowClient.class), + NAMESPACE, + TASK_QUEUE, + dataConverter, + new WorkerInterceptor[] {}); + handler.registerNexusServiceImplementations(new Object[] {serviceImpl}); + handler.start(); + return handler.handle(new NexusTask(task, null, null), metricsScope); + } + + private static PollNexusTaskQueueResponse.Builder startTask(Payload input) { + return PollNexusTaskQueueResponse.newBuilder() + .setRequest( + Request.newBuilder() + .setEndpoint(ENDPOINT) + .setStartOperation( + StartOperationRequest.newBuilder() + .setService(SERVICE) + .setOperation(OPERATION) + .setPayload(input))); + } + + private static String signature(NexusSerializationContext context) { + return context.getEndpoint() + ":" + context.getService() + ":" + context.getOperation(); + } + + @ServiceImpl(service = TestNexusServices.TestNexusService1.class) + public static class EchoServiceImpl { + @OperationImpl + public OperationHandler operation() { + return io.nexusrpc.handler.OperationHandler.sync( + (ctx, details, name) -> "Hello, " + name + "!"); + } + } + + @ServiceImpl(service = TestNexusServices.TestNexusService1.class) + public static class FailingServiceImpl { + @OperationImpl + public OperationHandler operation() { + return io.nexusrpc.handler.OperationHandler.sync( + (ctx, details, name) -> { + throw OperationException.failed(name); + }); + } + } + + /** + * Stamps the serialization context it was given onto every payload it encodes, and records each + * context it is handed so a test can assert which contexts were used and in what order. + */ + private static class SigningCodec implements PayloadCodec { + static final String SIGNATURE_KEY = "ser-ctx-signature"; + + private final List seen; + private final SerializationContext context; + + SigningCodec() { + this(Collections.synchronizedList(new ArrayList<>()), null); + } + + private SigningCodec(List seen, SerializationContext context) { + this.seen = seen; + this.context = context; + } + + List contexts() { + synchronized (seen) { + return new ArrayList<>(seen); + } + } + + @Override + @Nonnull + public PayloadCodec withContext(@Nonnull SerializationContext context) { + return new SigningCodec(seen, context); + } + + @Override + @Nonnull + public List encode(@Nonnull List payloads) { + seen.add(context); + if (!(context instanceof NexusSerializationContext)) { + return payloads; + } + List encoded = new ArrayList<>(payloads.size()); + for (Payload payload : payloads) { + encoded.add( + Payload.newBuilder(payload) + .putMetadata( + SIGNATURE_KEY, + ByteString.copyFromUtf8(signature((NexusSerializationContext) context))) + .build()); + } + return encoded; + } + + @Override + @Nonnull + public List decode(@Nonnull List payloads) { + seen.add(context); + if (!(context instanceof NexusSerializationContext)) { + return payloads; + } + String expected = signature((NexusSerializationContext) context); + List decoded = new ArrayList<>(payloads.size()); + for (Payload payload : payloads) { + ByteString actual = payload.getMetadataMap().get(SIGNATURE_KEY); + // Payloads encoded without a context stay readable, as the contract requires. + if (actual != null) { + Assert.assertEquals( + "payload should be decoded under the context it was encoded with", + expected, + actual.toStringUtf8()); + decoded.add(Payload.newBuilder(payload).removeMetadata(SIGNATURE_KEY).build()); + } else { + decoded.add(payload); + } + } + return decoded; + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/nexus/NexusSerializationContextTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/NexusSerializationContextTest.java new file mode 100644 index 0000000000..2a4daa8fc7 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/NexusSerializationContextTest.java @@ -0,0 +1,321 @@ +package io.temporal.workflow.nexus; + +import com.google.protobuf.ByteString; +import io.nexusrpc.OperationException; +import io.nexusrpc.handler.OperationHandler; +import io.nexusrpc.handler.OperationImpl; +import io.nexusrpc.handler.ServiceImpl; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.history.v1.HistoryEvent; +import io.temporal.api.nexus.v1.Endpoint; +import io.temporal.api.nexus.v1.EndpointSpec; +import io.temporal.api.nexus.v1.EndpointTarget; +import io.temporal.api.operatorservice.v1.CreateNexusEndpointRequest; +import io.temporal.api.operatorservice.v1.DeleteNexusEndpointRequest; +import io.temporal.client.WorkflowFailedException; +import io.temporal.client.WorkflowStub; +import io.temporal.common.converter.CodecDataConverter; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.payload.codec.PayloadCodec; +import io.temporal.payload.context.NexusSerializationContext; +import io.temporal.payload.context.SerializationContext; +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 io.temporal.workflow.shared.TestNexusServices; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import javax.annotation.Nonnull; +import org.junit.After; +import org.junit.Assert; +import org.junit.Assume; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; + +/** + * End-to-end coverage that a workflow calling a Nexus operation encodes its input, decodes its + * result, and converts its failures under the endpoint, service and operation of that operation. + * + *

Two endpoints are routed to the same worker so that a single workflow can call both and the + * payloads for each can be told apart on the wire. + * + *

Nexus requires a real server, so these are skipped unless {@code USE_EXTERNAL_SERVICE=true}. + */ +public class NexusSerializationContextTest { + private static final String RED_ENDPOINT = "red-nexus-endpoint"; + private static final String BLUE_ENDPOINT = "blue-nexus-endpoint"; + private static final String SERVICE = "TestNexusService1"; + private static final String OPERATION = "operation"; + + private static final SigningCodec CODEC = new SigningCodec(); + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(TwoEndpointWorkflowImpl.class, FailingWorkflowImpl.class) + .setNexusServiceImplementation(new TestNexusServiceImpl()) + .setWorkflowClientOptions( + io.temporal.client.WorkflowClientOptions.newBuilder() + .setDataConverter( + new CodecDataConverter( + DefaultDataConverter.STANDARD_INSTANCE, Collections.singletonList(CODEC))) + .build()) + .build(); + + private final List endpoints = new ArrayList<>(); + + @Before + public void setUp() { + Assume.assumeTrue( + "Nexus operations require a real server", SDKTestWorkflowRule.useExternalService); + CODEC.reset(); + endpoints.add(createEndpoint(RED_ENDPOINT)); + endpoints.add(createEndpoint(BLUE_ENDPOINT)); + } + + @After + public void tearDown() { + for (Endpoint endpoint : endpoints) { + testWorkflowRule + .getTestEnvironment() + .getOperatorServiceStubs() + .blockingStub() + .deleteNexusEndpoint( + DeleteNexusEndpointRequest.newBuilder() + .setId(endpoint.getId()) + .setVersion(endpoint.getVersion()) + .build()); + } + endpoints.clear(); + } + + @Test + public void inputAndResultUseTheOperationsOwnContext() { + TwoEndpointWorkflow workflow = + testWorkflowRule.newWorkflowStubTimeoutOptions(TwoEndpointWorkflow.class); + // Each operation must come back with the value it was called with, which only happens if the + // result was decoded under the same context it was encoded with. + Assert.assertEquals(Arrays.asList("Hello, red!", "Hello, blue!"), workflow.execute()); + String workflowId = WorkflowStub.fromTyped(workflow).getExecution().getWorkflowId(); + + // The payloads on the wire carry the context each operation was scheduled with. + Map inputSignatures = new HashMap<>(); + Map resultSignatures = new HashMap<>(); + Map scheduledEndpoints = new HashMap<>(); + for (HistoryEvent event : testWorkflowRule.getExecutionHistory(workflowId).getEvents()) { + if (event.hasNexusOperationScheduledEventAttributes()) { + io.temporal.api.history.v1.NexusOperationScheduledEventAttributes attrs = + event.getNexusOperationScheduledEventAttributes(); + scheduledEndpoints.put(event.getEventId(), attrs.getEndpoint()); + inputSignatures.put(attrs.getEndpoint(), signatureOf(attrs.getInput())); + } else if (event.hasNexusOperationCompletedEventAttributes()) { + io.temporal.api.history.v1.NexusOperationCompletedEventAttributes attrs = + event.getNexusOperationCompletedEventAttributes(); + resultSignatures.put( + scheduledEndpoints.get(attrs.getScheduledEventId()), signatureOf(attrs.getResult())); + } + } + + Assert.assertEquals( + "each operation's input should be encoded under its own endpoint", + expectedSignatures(), + inputSignatures); + Assert.assertEquals( + "each operation's result should be encoded under its own endpoint", + expectedSignatures(), + resultSignatures); + } + + @Test + public void failuresUseTheOperationsOwnContext() { + FailingWorkflow workflow = + testWorkflowRule.newWorkflowStubTimeoutOptions(FailingWorkflow.class); + Assert.assertThrows(WorkflowFailedException.class, workflow::execute); + + NexusSerializationContext expected = + new NexusSerializationContext(RED_ENDPOINT, SERVICE, OPERATION); + Assert.assertTrue( + "the caller should have converted the operation failure under the operation's context, " + + "but saw " + + CODEC.nexusContexts(), + CODEC.nexusContexts().contains(expected)); + } + + private Map expectedSignatures() { + Map expected = new HashMap<>(); + expected.put(RED_ENDPOINT, signature(RED_ENDPOINT)); + expected.put(BLUE_ENDPOINT, signature(BLUE_ENDPOINT)); + return expected; + } + + private static String signatureOf(Payload payload) { + ByteString signature = payload.getMetadataMap().get(SigningCodec.SIGNATURE_KEY); + return signature == null ? null : signature.toStringUtf8(); + } + + private static String signature(String endpoint) { + return endpoint + ":" + SERVICE + ":" + OPERATION; + } + + private Endpoint createEndpoint(String name) { + return testWorkflowRule + .getTestEnvironment() + .getOperatorServiceStubs() + .blockingStub() + .createNexusEndpoint( + CreateNexusEndpointRequest.newBuilder() + .setSpec( + EndpointSpec.newBuilder() + .setName(name) + .setTarget( + EndpointTarget.newBuilder() + .setWorker( + EndpointTarget.Worker.newBuilder() + .setNamespace( + testWorkflowRule.getTestEnvironment().getNamespace()) + .setTaskQueue(testWorkflowRule.getTaskQueue())))) + .build()) + .getEndpoint(); + } + + @WorkflowInterface + public interface TwoEndpointWorkflow { + @WorkflowMethod + List execute(); + } + + @WorkflowInterface + public interface FailingWorkflow { + @WorkflowMethod + void execute(); + } + + public static class TwoEndpointWorkflowImpl implements TwoEndpointWorkflow { + @Override + public List execute() { + return Arrays.asList( + stubFor(RED_ENDPOINT).operation("red"), stubFor(BLUE_ENDPOINT).operation("blue")); + } + } + + public static class FailingWorkflowImpl implements FailingWorkflow { + @Override + public void execute() { + stubFor(RED_ENDPOINT).operation("fail"); + } + } + + private static TestNexusServices.TestNexusService1 stubFor(String endpoint) { + return Workflow.newNexusServiceStub( + TestNexusServices.TestNexusService1.class, + NexusServiceOptions.newBuilder() + .setEndpoint(endpoint) + .setOperationOptions( + NexusOperationOptions.newBuilder() + .setScheduleToCloseTimeout(Duration.ofSeconds(20)) + .build()) + .build()); + } + + @ServiceImpl(service = TestNexusServices.TestNexusService1.class) + public static class TestNexusServiceImpl { + @OperationImpl + public OperationHandler operation() { + return OperationHandler.sync( + (ctx, details, name) -> { + if ("fail".equals(name)) { + throw OperationException.failed("operation failed on purpose"); + } + return "Hello, " + name + "!"; + }); + } + } + + /** + * Stamps the Nexus context it was given onto every payload it encodes, so the context used for a + * payload can be read back off the wire, and records the Nexus contexts it was handed. + */ + private static class SigningCodec implements PayloadCodec { + static final String SIGNATURE_KEY = "ser-ctx-signature"; + + // Shared by every instance derived via withContext, so a test sees all contexts that were used. + private final List seen; + private final SerializationContext context; + + SigningCodec() { + this(Collections.synchronizedList(new ArrayList<>()), null); + } + + private SigningCodec(List seen, SerializationContext context) { + this.seen = seen; + this.context = context; + } + + void reset() { + seen.clear(); + } + + List nexusContexts() { + List result = new ArrayList<>(); + synchronized (seen) { + for (SerializationContext each : seen) { + if (each instanceof NexusSerializationContext) { + result.add((NexusSerializationContext) each); + } + } + } + return result; + } + + @Override + @Nonnull + public PayloadCodec withContext(@Nonnull SerializationContext context) { + return new SigningCodec(seen, context); + } + + @Override + @Nonnull + public List encode(@Nonnull List payloads) { + seen.add(context); + if (!(context instanceof NexusSerializationContext)) { + return payloads; + } + NexusSerializationContext nexus = (NexusSerializationContext) context; + String signature = + nexus.getEndpoint() + ":" + nexus.getService() + ":" + nexus.getOperation(); + List encoded = new ArrayList<>(payloads.size()); + for (Payload payload : payloads) { + encoded.add( + Payload.newBuilder(payload) + .putMetadata(SIGNATURE_KEY, ByteString.copyFromUtf8(signature)) + .build()); + } + return encoded; + } + + @Override + @Nonnull + public List decode(@Nonnull List payloads) { + seen.add(context); + List decoded = new ArrayList<>(payloads.size()); + for (Payload payload : payloads) { + // Payloads encoded without a context stay readable, as the contract requires. + if (payload.getMetadataMap().containsKey(SIGNATURE_KEY)) { + decoded.add(Payload.newBuilder(payload).removeMetadata(SIGNATURE_KEY).build()); + } else { + decoded.add(payload); + } + } + return decoded; + } + } +}