From 7e523513184437d158dd65443c71fddc3b2c1dd6 Mon Sep 17 00:00:00 2001 From: Joshua Frenchwood Date: Tue, 27 Oct 2026 13:55:08 -0500 Subject: [PATCH] WorkflowUpdate nexusserializationcontext --- CHANGELOG.md | 8 +- .../client/RootWorkflowClientInvoker.java | 76 ++++-- .../PropagatedNexusSerializationContext.java | 9 +- .../internal/replay/ReplayWorkflow.java | 11 +- .../replay/ReplayWorkflowExecutor.java | 5 + .../temporal/internal/sync/SyncWorkflow.java | 15 +- .../internal/sync/SyncWorkflowContext.java | 29 ++- .../internal/sync/UpdateDispatcher.java | 13 +- .../sync/WorkflowExecutionHandler.java | 39 ++-- .../context/NexusSerializationContext.java | 8 +- ...tInvokerNexusSerializationContextTest.java | 188 ++++++++++++++- ...eplayWorkflowRunTaskHandlerCacheTests.java | 2 + ...owUpdateNexusSerializationContextTest.java | 217 ++++++++++++++++++ ...usSerializationContextPropagationTest.java | 118 +++++++++- 14 files changed, 666 insertions(+), 72 deletions(-) create mode 100644 temporal-sdk/src/test/java/io/temporal/internal/sync/WorkflowUpdateNexusSerializationContextTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 8a30365d81..a390056d46 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -51,10 +51,10 @@ 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. +- Nexus handlers now propagate `NexusSerializationContext` to the workflows, workflow updates, 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)`. 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 e2284aad2e..147a7bb7a9 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 @@ -118,6 +118,27 @@ private DataConverter workflowConverter( return converterFactory.forWorkflow(workflowId, runId, workflowType); } + /** + * The data converter for a workflow's or an update's result and failure. A workflow or an update + * started by a Nexus operation handler encodes them with the handler's Nexus serialization + * context, which the server returns with them. Otherwise they use the workflow context. + * + * @param propagated the context returned by the server, or null if it returned none + * @param workflowConverter the converter to use when the server returned no context + */ + private DataConverter outcomeConverter( + @Nullable PropagatedSerializationContext propagated, + WorkflowExecution execution, + DataConverter workflowConverter) { + return propagated != null + ? converterFactory.forWorkflow( + PropagatedNexusSerializationContext.fromProto(propagated), + execution.getWorkflowId(), + execution.getRunId(), + null) + : workflowConverter; + } + private void storeHeader( Header.Builder header, String workflowId, @@ -475,14 +496,12 @@ 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); + outcomeConverter( + closed.hasPropagatedNexusSerializationContext() + ? closed.getPropagatedNexusSerializationContext() + : null, + execution, + workflowConverter(execution)); // The long poll helpers return a response whose single event is the close event. HistoryEvent closeEvent = closed.getHistory().getEvents(0); Optional resultValue = @@ -615,6 +634,17 @@ private UpdateWorkflowExecutionRequest toUpdateWorkflowExecutionRequest( .setMeta(Meta.newBuilder().setUpdateId(input.getUpdateId()).setIdentity(identity)) .setInput(updateInput); + // An update sent by 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 worker and to clients reading the outcome. The update's + // arguments keep the workflow context. + @Nullable + PropagatedSerializationContext propagatedNexusSerializationContext = + PropagatedNexusSerializationContext.current(); + if (propagatedNexusSerializationContext != null) { + requestBuilder.setPropagatedNexusSerializationContext(propagatedNexusSerializationContext); + } + // If this update is being issued via TemporalNexusClientImpl.startWorkflowUpdate, // set the fields the server needs to deliver the Nexus completion callback if (CurrentNexusOperationContext.isNexusContext()) { @@ -666,15 +696,19 @@ private WorkflowUpdateHandle toUpdateHandle( UpdateWorkflowExecutionResponse result, DataConverter dataConverterWithWorkflowContext) { if (result.hasOutcome()) { + DataConverter dataConverter = + outcomeConverter( + result.hasPropagatedNexusSerializationContext() + ? result.getPropagatedNexusSerializationContext() + : null, + result.getUpdateRef().getWorkflowExecution(), + dataConverterWithWorkflowContext); switch (result.getOutcome().getValueCase()) { case SUCCESS: Optional updateResult = Optional.of(result.getOutcome().getSuccess()); R resultValue = convertResultPayloads( - updateResult, - input.getResultClass(), - input.getResultType(), - dataConverterWithWorkflowContext); + updateResult, input.getResultClass(), input.getResultType(), dataConverter); return new CompletedWorkflowUpdateHandleImpl<>( result.getUpdateRef().getUpdateId(), result.getUpdateRef().getWorkflowExecution(), @@ -687,8 +721,7 @@ private WorkflowUpdateHandle toUpdateHandle( result.getUpdateRef().getWorkflowExecution(), result.getUpdateRef().getUpdateId(), input.getUpdateName(), - dataConverterWithWorkflowContext.failureToException( - result.getOutcome().getFailure()))); + dataConverter.failureToException(result.getOutcome().getFailure()))); default: throw new RuntimeException( "Received unexpected outcome from update request: " @@ -748,21 +781,24 @@ public PollWorkflowUpdateOutput pollWorkflowUpdate(PollWorkflowUpdateInpu future.thenApply( (result) -> { if (result.hasOutcome()) { + DataConverter dataConverter = + outcomeConverter( + result.hasPropagatedNexusSerializationContext() + ? result.getPropagatedNexusSerializationContext() + : null, + input.getWorkflowExecution(), + dataConverterWithWorkflowContext); switch (result.getOutcome().getValueCase()) { case SUCCESS: Optional updateResult = Optional.of(result.getOutcome().getSuccess()); return convertResultPayloads( - updateResult, - input.getResultClass(), - input.getResultType(), - dataConverterWithWorkflowContext); + updateResult, input.getResultClass(), input.getResultType(), dataConverter); case FAILURE: throw new WorkflowUpdateException( input.getWorkflowExecution(), input.getUpdateId(), input.getUpdateName(), - dataConverterWithWorkflowContext.failureToException( - result.getOutcome().getFailure())); + dataConverter.failureToException(result.getOutcome().getFailure())); default: throw new RuntimeException( "Received unexpected outcome from poll update request: " 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 index 9690a6a925..9b2b997d36 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/nexus/PropagatedNexusSerializationContext.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/nexus/PropagatedNexusSerializationContext.java @@ -9,10 +9,11 @@ * 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. + *

When a Nexus handler starts a workflow, sends a workflow update or starts a standalone + * activity, the SDK sends the handler's serialization context with the 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 { diff --git a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflow.java b/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflow.java index b66b7cfc02..b9464b404e 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflow.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflow.java @@ -5,7 +5,9 @@ import io.temporal.api.history.v1.HistoryEvent; import io.temporal.api.query.v1.WorkflowQuery; import io.temporal.internal.statemachines.UpdateProtocolCallback; +import io.temporal.payload.context.NexusSerializationContext; import java.util.Optional; +import javax.annotation.Nullable; /** * Manages event loop, workflow method, an abstraction level over Deterministic Runner to provide a @@ -18,13 +20,20 @@ public interface ReplayWorkflow { /** Handle an external signal event. */ void handleSignal(String signalName, Optional input, long eventId, Header header); - /** Handle an update workflow execution event */ + /** + * Handle an update workflow execution event. + * + * @param nexusSerializationContext the serialization context of the Nexus operation handler that + * sent the update, or null if it was not sent by one. When set, the update's result, failure + * and rejection, which are returned to the Nexus caller, are serialized with it. + */ void handleUpdate( String updateName, String updateId, Optional input, long eventId, Header header, + @Nullable NexusSerializationContext nexusSerializationContext, UpdateProtocolCallback callbacks); /** diff --git a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowExecutor.java b/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowExecutor.java index 9efe315c70..0dc3ad2696 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowExecutor.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/replay/ReplayWorkflowExecutor.java @@ -17,6 +17,7 @@ import io.temporal.internal.common.UpdateMessage; import io.temporal.internal.logging.LoggerTag; import io.temporal.internal.logging.PrefixedMdc; +import io.temporal.internal.nexus.PropagatedNexusSerializationContext; import io.temporal.internal.statemachines.WorkflowStateMachines; import io.temporal.internal.sync.SignalHandlerInfo; import io.temporal.internal.sync.UpdateHandlerInfo; @@ -194,6 +195,10 @@ public void handleWorkflowExecutionUpdated(UpdateMessage updateMessage) { args, protocolMessage.getEventId(), input.getHeader(), + update.hasPropagatedNexusSerializationContext() + ? PropagatedNexusSerializationContext.fromProto( + update.getPropagatedNexusSerializationContext()) + : null, updateMessage.getCallbacks()); } catch (InvalidProtocolBufferException e) { throw new IllegalStateException("Message is not an update."); 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 d8c7cc33fe..fdb5a1d4f5 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 @@ -19,6 +19,7 @@ import io.temporal.internal.statemachines.UpdateProtocolCallback; import io.temporal.internal.worker.WorkflowExecutionException; import io.temporal.internal.worker.WorkflowExecutorCache; +import io.temporal.payload.context.NexusSerializationContext; import io.temporal.payload.context.WorkflowSerializationContext; import io.temporal.worker.WorkflowImplementationOptions; import io.temporal.workflow.UpdateInfo; @@ -151,8 +152,11 @@ public void handleUpdate( Optional input, long eventId, Header header, + @Nullable NexusSerializationContext nexusSerializationContext, UpdateProtocolCallback callbacks) { final UpdateInfo updateInfo = new UpdateInfoImpl(updateName, updateId); + DataConverter outcomeDataConverter = + workflowContext.getDataConverterForUpdateOutcome(nexusSerializationContext); runner.executeInWorkflowThread( "update " + updateName, () -> { @@ -165,7 +169,8 @@ public void handleUpdate( if (!callbacks.isReplaying()) { try { workflowContext.setReadOnly(true); - workflowProc.handleValidateUpdate(updateName, updateId, input, eventId, header); + workflowProc.handleValidateUpdate( + updateName, updateId, input, eventId, header, outcomeDataConverter); } catch (ReadOnlyException r) { // Rethrow instead on rejecting the update to fail the WFT throw r; @@ -173,10 +178,7 @@ public void handleUpdate( callbacks.reject(e.getFailure()); return; } catch (Exception e) { - callbacks.reject( - workflowContext - .getDataConverterWithCurrentWorkflowContext() - .exceptionToFailure(e)); + callbacks.reject(outcomeDataConverter.exceptionToFailure(e)); return; } finally { workflowContext.setReadOnly(false); @@ -185,7 +187,8 @@ public void handleUpdate( callbacks.accept(); try { Optional result = - workflowProc.handleExecuteUpdate(updateName, updateId, input, eventId, header); + workflowProc.handleExecuteUpdate( + updateName, updateId, input, eventId, header, outcomeDataConverter); callbacks.complete(result, null); } catch (WorkflowExecutionException e) { callbacks.complete(Optional.empty(), e.getFailure()); 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 3f1932f42c..eb99bdc7a0 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 @@ -352,8 +352,14 @@ public void handleValidateUpdate( } public Optional handleExecuteUpdate( - String updateName, String updateId, Optional input, long eventId, Header header) { - return updateDispatcher.handleExecuteUpdate(updateName, updateId, input, eventId, header); + String updateName, + String updateId, + Optional input, + long eventId, + Header header, + DataConverter resultDataConverter) { + return updateDispatcher.handleExecuteUpdate( + updateName, updateId, input, eventId, header, resultDataConverter); } public void handleInterceptedValidateUpdate(WorkflowInboundCallsInterceptor.UpdateInput input) { @@ -1319,8 +1325,8 @@ void setPropagatedNexusSerializationContext( /** * 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. + * Nexus operation handler, otherwise the workflow context. Input, memo, signals and queries + * always use the workflow context. Updates use {@link #getDataConverterForUpdateOutcome}. */ DataConverter getDataConverterForResult() { return dataConverterWithPropagatedNexusContext != null @@ -1657,12 +1663,17 @@ public Failure mapWorkflowExceptionToFailure(Throwable 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. + * The data converter for an update's result, failure and rejection, which are returned to the + * update's caller: the Nexus context of the Nexus operation handler that sent the update, if it + * was sent by one, otherwise the workflow context. Unlike {@link #getDataConverterForResult}, + * this doesn't depend on how the workflow was started, because the outcome belongs to the update + * rather than closing the workflow. The update's arguments always use the workflow context. */ - Failure mapUpdateExceptionToFailure(Throwable failure) { - return dataConverterWithCurrentWorkflowContext.exceptionToFailure(failure); + DataConverter getDataConverterForUpdateOutcome( + @Nullable NexusSerializationContext nexusSerializationContext) { + return nexusSerializationContext != null + ? dataConverter.withContext(nexusSerializationContext) + : dataConverterWithCurrentWorkflowContext; } @Nullable diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/UpdateDispatcher.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/UpdateDispatcher.java index 8ce53e87a5..500b3eae34 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/UpdateDispatcher.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/UpdateDispatcher.java @@ -71,8 +71,17 @@ public void handleValidateUpdate( } } + /** + * @param resultDataConverter the converter for the update's result, which may differ from the + * workflow converter used for its arguments + */ public Optional handleExecuteUpdate( - String updateName, String updateId, Optional input, long eventId, Header header) { + String updateName, + String updateId, + Optional input, + long eventId, + Header header, + DataConverter resultDataConverter) { WorkflowOutboundCallsInterceptor.UpdateRegistrationRequest handler = updateCallbacks.get(updateName); Object[] args; @@ -99,7 +108,7 @@ public Optional handleExecuteUpdate( .executeUpdate( new WorkflowInboundCallsInterceptor.UpdateInput(updateName, header, args)) .getResult(); - return dataConverterWithWorkflowContext.toPayloads(result); + return resultDataConverter.toPayloads(result); } catch (DestroyWorkflowThreadError e) { threadDestroyed = true; throw e; 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 5ab392bc7b..ddebca88a4 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 @@ -5,6 +5,7 @@ import io.temporal.api.common.v1.Payloads; import io.temporal.api.history.v1.WorkflowExecutionStartedEventAttributes; +import io.temporal.common.converter.DataConverter; import io.temporal.common.interceptors.Header; import io.temporal.failure.CanceledFailure; import io.temporal.failure.TemporalFailure; @@ -101,11 +102,12 @@ public void handleValidateUpdate( String updateId, Optional input, long eventId, - io.temporal.api.common.v1.Header header) { + io.temporal.api.common.v1.Header header, + DataConverter outcomeDataConverter) { try { context.handleValidateUpdate(updateName, updateId, input, eventId, new Header(header)); } catch (Throwable e) { - applyWorkflowFailurePolicyAndRethrow(e, true); + applyWorkflowFailurePolicyAndRethrow(e, outcomeDataConverter); } } @@ -114,27 +116,31 @@ public Optional handleExecuteUpdate( String updateId, Optional input, long eventId, - io.temporal.api.common.v1.Header header) { + io.temporal.api.common.v1.Header header, + DataConverter outcomeDataConverter) { try { - return context.handleExecuteUpdate(updateName, updateId, input, eventId, new Header(header)); + return context.handleExecuteUpdate( + updateName, updateId, input, eventId, new Header(header), outcomeDataConverter); } catch (UnsupportedContinueAsNewRequest e) { // Re-throw to fail the workflow task throw e; } catch (Throwable e) { - applyWorkflowFailurePolicyAndRethrow(e, true); + applyWorkflowFailurePolicyAndRethrow(e, outcomeDataConverter); } return Optional.empty(); } private void applyWorkflowFailurePolicyAndRethrow(Throwable e) { - applyWorkflowFailurePolicyAndRethrow(e, false); + applyWorkflowFailurePolicyAndRethrow(e, null); } /** - * @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 + * @param updateOutcomeDataConverter the converter for the update's outcome if 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, or null otherwise */ - private void applyWorkflowFailurePolicyAndRethrow(Throwable e, boolean forUpdate) { + private void applyWorkflowFailurePolicyAndRethrow( + Throwable e, @Nullable DataConverter updateOutcomeDataConverter) { if (e instanceof DestroyWorkflowThreadError) { throw (DestroyWorkflowThreadError) e; } @@ -142,18 +148,19 @@ private void applyWorkflowFailurePolicyAndRethrow(Throwable e, boolean forUpdate Class[] failTypes = implementationOptions.getFailWorkflowExceptionTypes(); if (exception instanceof TemporalFailure) { - throwAndFailWorkflowExecution(exception, forUpdate); + throwAndFailWorkflowExecution(exception, updateOutcomeDataConverter); } for (Class failType : failTypes) { if (failType.isAssignableFrom(exception.getClass())) { - throwAndFailWorkflowExecution(exception, forUpdate); + throwAndFailWorkflowExecution(exception, updateOutcomeDataConverter); } } throw wrap(exception); } - private void throwAndFailWorkflowExecution(Throwable exception, boolean forUpdate) { + private void throwAndFailWorkflowExecution( + Throwable exception, @Nullable DataConverter updateOutcomeDataConverter) { ReplayWorkflowContext replayWorkflowContext = context.getReplayContext(); @Nullable String fullReplayDirectQueryName = replayWorkflowContext.getFullReplayDirectQueryName(); @@ -182,11 +189,11 @@ private void throwAndFailWorkflowExecution(Throwable exception, boolean forUpdat } } - // 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. + // An update failure belongs to the update, so it uses the update's own converter rather than + // the one for the failure that closes the workflow. throw new WorkflowExecutionException( - forUpdate - ? context.mapUpdateExceptionToFailure(exception) + updateOutcomeDataConverter != null + ? updateOutcomeDataConverter.exceptionToFailure(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 8c46e23fd3..27bbacf72c 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 @@ -10,10 +10,10 @@ * service, and resolved operation the payload belongs to. * *

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. + * inputs, results, and failures. When a handler starts a workflow or standalone activity, or sends + * a workflow update, 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. * *

To complete such an activity asynchronously, store the context from {@link * io.temporal.activity.ActivityInfo#getNexusSerializationContext()} with the task token and pass it 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 index 6896e7dbf5..f5e66881de 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerNexusSerializationContextTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerNexusSerializationContextTest.java @@ -14,6 +14,7 @@ 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.enums.v1.UpdateWorkflowExecutionLifecycleStage; import io.temporal.api.failure.v1.ApplicationFailureInfo; import io.temporal.api.failure.v1.Failure; import io.temporal.api.history.v1.History; @@ -22,18 +23,30 @@ 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.update.v1.Outcome; +import io.temporal.api.update.v1.Request; +import io.temporal.api.update.v1.UpdateRef; +import io.temporal.api.update.v1.WaitPolicy; import io.temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryRequest; import io.temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse; +import io.temporal.api.workflowservice.v1.PollWorkflowExecutionUpdateRequest; +import io.temporal.api.workflowservice.v1.PollWorkflowExecutionUpdateResponse; 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.api.workflowservice.v1.UpdateWorkflowExecutionRequest; +import io.temporal.api.workflowservice.v1.UpdateWorkflowExecutionResponse; import io.temporal.client.WorkflowClient; import io.temporal.client.WorkflowFailedException; import io.temporal.client.WorkflowOptions; +import io.temporal.client.WorkflowUpdateException; +import io.temporal.client.WorkflowUpdateHandle; 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.PollWorkflowUpdateInput; +import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.StartUpdateInput; import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowSignalWithStartInput; import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowStartInput; import io.temporal.internal.client.external.GenericWorkflowClient; @@ -45,6 +58,7 @@ import java.util.Collections; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; import java.util.concurrent.TimeUnit; import org.junit.After; import org.junit.Assert; @@ -53,8 +67,9 @@ 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. + * Unit tests for propagating the Nexus serialization context to workflows started, and updates + * sent, by a Nexus operation handler, and for decoding their results with the context the server + * returns. */ public class RootWorkflowClientInvokerNexusSerializationContextTest { @@ -214,6 +229,92 @@ public void getResultAsyncKeepsContextAcrossEmptyPolls() throws Exception { Assert.assertEquals(Collections.singletonList(NEXUS), codec.decodeContexts()); } + @Test + public void updateFromNexusHandlerPropagatesContextButKeepsWorkflowContextForArguments() + throws Exception { + enterNexusHandler(NEXUS_CONTEXT); + when(genericClient.update(any(UpdateWorkflowExecutionRequest.class), any(Deadline.class))) + .thenReturn(completedUpdate(successOutcome(NEXUS_CONTEXT), PROPAGATED)); + + String result = invoker.startUpdate(newStartUpdateInput()).getResult(); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(UpdateWorkflowExecutionRequest.class); + verify(genericClient).update(captor.capture(), any(Deadline.class)); + Request request = captor.getValue().getRequest(); + Assert.assertEquals(PROPAGATED, request.getPropagatedNexusSerializationContext()); + // The update's arguments are not returned to the Nexus caller, so they keep the workflow + // context. + Assert.assertEquals(WORKFLOW, contextOf(request.getInput().getArgs().getPayloads(0))); + Assert.assertEquals("result", result); + Assert.assertEquals(Collections.singletonList(NEXUS), codec.decodeContexts()); + } + + @Test + public void updateOutsideNexusHandlerDoesNotPropagateContext() throws Exception { + when(genericClient.update(any(UpdateWorkflowExecutionRequest.class), any(Deadline.class))) + .thenReturn(completedUpdate(successOutcome(null), null)); + + invoker.startUpdate(newStartUpdateInput()).getResult(); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(UpdateWorkflowExecutionRequest.class); + verify(genericClient).update(captor.capture(), any(Deadline.class)); + Assert.assertFalse(captor.getValue().getRequest().hasPropagatedNexusSerializationContext()); + Assert.assertEquals(Collections.singletonList(WORKFLOW), codec.decodeContexts()); + } + + @Test + public void completedUpdateDecodesFailureWithPropagatedContext() { + when(genericClient.update(any(UpdateWorkflowExecutionRequest.class), any(Deadline.class))) + .thenReturn(completedUpdate(failureOutcome(NEXUS_CONTEXT), PROPAGATED)); + + WorkflowUpdateHandle handle = invoker.startUpdate(newStartUpdateInput()); + + Assert.assertThrows(WorkflowUpdateException.class, handle::getResult); + Assert.assertEquals(Collections.singletonList(NEXUS), codec.decodeContexts()); + } + + @Test + public void pollUpdateDecodesWithPropagatedContext() throws Exception { + when(genericClient.pollUpdateAsync( + any(PollWorkflowExecutionUpdateRequest.class), any(Deadline.class))) + .thenReturn( + CompletableFuture.completedFuture( + polledUpdate(successOutcome(NEXUS_CONTEXT), PROPAGATED))); + + String result = invoker.pollWorkflowUpdate(newPollUpdateInput()).getResult().get(); + + Assert.assertEquals("result", result); + Assert.assertEquals(Collections.singletonList(NEXUS), codec.decodeContexts()); + } + + @Test + public void pollUpdateWithoutPropagatedContextUsesWorkflowContext() throws Exception { + when(genericClient.pollUpdateAsync( + any(PollWorkflowExecutionUpdateRequest.class), any(Deadline.class))) + .thenReturn(CompletableFuture.completedFuture(polledUpdate(successOutcome(null), null))); + + invoker.pollWorkflowUpdate(newPollUpdateInput()).getResult().get(); + + Assert.assertEquals(Collections.singletonList(WORKFLOW), codec.decodeContexts()); + } + + @Test + public void pollUpdateDecodesFailureWithPropagatedContext() { + when(genericClient.pollUpdateAsync( + any(PollWorkflowExecutionUpdateRequest.class), any(Deadline.class))) + .thenReturn( + CompletableFuture.completedFuture( + polledUpdate(failureOutcome(NEXUS_CONTEXT), PROPAGATED))); + + CompletableFuture result = invoker.pollWorkflowUpdate(newPollUpdateInput()).getResult(); + + CompletionException e = Assert.assertThrows(CompletionException.class, result::join); + Assert.assertTrue(e.getCause() instanceof WorkflowUpdateException); + Assert.assertEquals(Collections.singletonList(NEXUS), codec.decodeContexts()); + } + private void enterNexusHandler(NexusSerializationContext serializationContext) { InternalNexusOperationContext nexusContext = new InternalNexusOperationContext( @@ -302,4 +403,87 @@ private static Payloads payloads(String value, NexusSerializationContext encoded : stamp(payload, encodedWith)) .build(); } + + private static StartUpdateInput newStartUpdateInput() { + return new StartUpdateInput<>( + WorkflowExecution.newBuilder().setWorkflowId(WORKFLOW_ID).build(), + Optional.of("TestWorkflow"), + "update", + Header.empty(), + "update-id", + new Object[] {"input"}, + String.class, + String.class, + "", + WaitPolicy.newBuilder() + .setLifecycleStage( + UpdateWorkflowExecutionLifecycleStage + .UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_COMPLETED) + .build()); + } + + private static PollWorkflowUpdateInput newPollUpdateInput() { + return new PollWorkflowUpdateInput<>( + WorkflowExecution.newBuilder().setWorkflowId(WORKFLOW_ID).setRunId("run-1").build(), + "update", + "update-id", + String.class, + String.class, + 10, + TimeUnit.SECONDS); + } + + private static UpdateWorkflowExecutionResponse completedUpdate( + Outcome outcome, PropagatedSerializationContext propagated) { + UpdateWorkflowExecutionResponse.Builder response = + UpdateWorkflowExecutionResponse.newBuilder() + .setUpdateRef(updateRef()) + .setStage( + UpdateWorkflowExecutionLifecycleStage + .UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_COMPLETED) + .setOutcome(outcome); + if (propagated != null) { + response.setPropagatedNexusSerializationContext(propagated); + } + return response.build(); + } + + private static PollWorkflowExecutionUpdateResponse polledUpdate( + Outcome outcome, PropagatedSerializationContext propagated) { + PollWorkflowExecutionUpdateResponse.Builder response = + PollWorkflowExecutionUpdateResponse.newBuilder() + .setUpdateRef(updateRef()) + .setStage( + UpdateWorkflowExecutionLifecycleStage + .UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_COMPLETED) + .setOutcome(outcome); + if (propagated != null) { + response.setPropagatedNexusSerializationContext(propagated); + } + return response.build(); + } + + private static UpdateRef updateRef() { + return UpdateRef.newBuilder() + .setWorkflowExecution( + WorkflowExecution.newBuilder().setWorkflowId(WORKFLOW_ID).setRunId("run-1")) + .setUpdateId("update-id") + .build(); + } + + private static Outcome successOutcome(NexusSerializationContext encodedWith) { + return Outcome.newBuilder().setSuccess(payloads("result", encodedWith)).build(); + } + + private static Outcome failureOutcome(NexusSerializationContext encodedWith) { + return Outcome.newBuilder() + .setFailure( + Failure.newBuilder() + .setMessage("failed") + .setApplicationFailureInfo( + ApplicationFailureInfo.newBuilder() + .setType("TestFailure") + .setDetails(payloads("detail", encodedWith)))) + .build(); + } } diff --git a/temporal-sdk/src/test/java/io/temporal/internal/replay/ReplayWorkflowRunTaskHandlerCacheTests.java b/temporal-sdk/src/test/java/io/temporal/internal/replay/ReplayWorkflowRunTaskHandlerCacheTests.java index dfb576a5fb..4f51c83887 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/replay/ReplayWorkflowRunTaskHandlerCacheTests.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/replay/ReplayWorkflowRunTaskHandlerCacheTests.java @@ -24,6 +24,7 @@ import io.temporal.internal.worker.SingleWorkerOptions; import io.temporal.internal.worker.WorkflowExecutorCache; import io.temporal.internal.worker.WorkflowRunLockManager; +import io.temporal.payload.context.NexusSerializationContext; import io.temporal.serviceclient.MetricsTag; import io.temporal.testUtils.HistoryUtils; import io.temporal.testing.CloudTestExclusion.RequiresLocalServer; @@ -290,6 +291,7 @@ public void handleUpdate( Optional input, long eventId, Header header, + NexusSerializationContext nexusSerializationContext, UpdateProtocolCallback callbacks) {} @Override diff --git a/temporal-sdk/src/test/java/io/temporal/internal/sync/WorkflowUpdateNexusSerializationContextTest.java b/temporal-sdk/src/test/java/io/temporal/internal/sync/WorkflowUpdateNexusSerializationContextTest.java new file mode 100644 index 0000000000..c237051d73 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/sync/WorkflowUpdateNexusSerializationContextTest.java @@ -0,0 +1,217 @@ +package io.temporal.internal.sync; + +import static org.mockito.Mockito.mock; + +import com.uber.m3.tally.NoopScope; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowClientOptions; +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.internal.nexus.CurrentNexusOperationContext; +import io.temporal.internal.nexus.InternalNexusOperationContext; +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.UpdateValidatorMethod; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; +import java.util.Arrays; +import java.util.UUID; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +/** + * Tests that the worker encodes the result, failure and rejection of an update sent by a Nexus + * operation handler, which are returned to the Nexus caller, with the Nexus context carried on the + * update request, and keeps the workflow context for the update's arguments. + * + *

The test server passes the update request, including the context, through to the worker, so + * each test sends the update from inside a Nexus operation context on the test thread. + */ +@CloudTestExclusionNote("This test directly creates and controls a local test service.") +@Category(RequiresLocalServer.class) +public class WorkflowUpdateNexusSerializationContextTest { + + private static final String TASK_QUEUE = "nexus-serialization-context-update"; + private static final String NAMESPACE = "default"; + private static final NexusSerializationContext NEXUS_CONTEXT = + new NexusSerializationContext("endpoint", "service", "operation"); + private static final String NEXUS = ContextRecordingCodec.describe(NEXUS_CONTEXT); + + @WorkflowInterface + public interface UpdateTargetWorkflow { + @WorkflowMethod + void execute(); + + @UpdateMethod + String echo(String input); + + @UpdateValidatorMethod(updateName = "echo") + void validateEcho(String input); + + @UpdateMethod + void finish(); + } + + public static class UpdateTargetWorkflowImpl implements UpdateTargetWorkflow { + private boolean finished; + + @Override + public void execute() { + Workflow.await(() -> finished); + } + + @Override + public String echo(String input) { + if ("fail".equals(input)) { + throw ApplicationFailure.newNonRetryableFailure("failed", "TestFailure", "detail"); + } + return input; + } + + @Override + public void validateEcho(String input) { + if ("reject".equals(input)) { + throw ApplicationFailure.newNonRetryableFailure("rejected", "TestFailure", "detail"); + } + } + + @Override + public void finish() { + finished = true; + } + } + + private ContextRecordingCodec codec; + private TestWorkflowEnvironment testEnvironment; + private Worker worker; + private UpdateTargetWorkflow workflow; + private String workflowId; + + @Before + public void setUp() { + codec = new ContextRecordingCodec(); + testEnvironment = + TestWorkflowEnvironment.newInstance( + TestEnvironmentOptions.newBuilder() + // With time skipping, updates aren'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(UpdateTargetWorkflowImpl.class); + testEnvironment.start(); + + workflowId = "nexus-serialization-context-update-" + UUID.randomUUID(); + workflow = + testEnvironment + .getWorkflowClient() + .newWorkflowStub( + UpdateTargetWorkflow.class, + WorkflowOptions.newBuilder() + .setTaskQueue(TASK_QUEUE) + .setWorkflowId(workflowId) + .build()); + WorkflowClient.start(workflow::execute); + } + + @After + public void tearDown() { + CurrentNexusOperationContext.unset(); + testEnvironment.close(); + } + + @Test + public void updateSentByNexusHandlerUsesNexusContextForResult() throws Exception { + codec.reset(); + Assert.assertEquals("hello", sendFromNexusHandler("hello")); + + // The client encodes the argument and the worker decodes it with the workflow context; the + // worker encodes the result with the Nexus context. + Assert.assertEquals(Arrays.asList(workflowContext(), NEXUS), codec.encodeContexts()); + Assert.assertEquals(workflowContext(), codec.decodeContexts().get(0)); + + // Replaying re-encodes the result with the context from the update's accepted event. The + // finish update wasn't sent by a Nexus handler, so its result keeps the workflow context. + WorkflowExecutionHistory history = finishAndFetchHistory(); + codec.reset(); + worker.replayWorkflowExecution(history); + Assert.assertEquals(Arrays.asList(NEXUS, workflowContext()), codec.encodeContexts()); + } + + @Test + public void updateSentByNexusHandlerUsesNexusContextForFailure() throws Exception { + codec.reset(); + Assert.assertThrows(WorkflowUpdateException.class, () -> sendFromNexusHandler("fail")); + + Assert.assertEquals(Arrays.asList(workflowContext(), NEXUS), codec.encodeContexts()); + + WorkflowExecutionHistory history = finishAndFetchHistory(); + codec.reset(); + worker.replayWorkflowExecution(history); + Assert.assertEquals(Arrays.asList(NEXUS, workflowContext()), codec.encodeContexts()); + } + + @Test + public void updateSentByNexusHandlerUsesNexusContextForRejection() { + codec.reset(); + Assert.assertThrows(WorkflowUpdateException.class, () -> sendFromNexusHandler("reject")); + + Assert.assertEquals(Arrays.asList(workflowContext(), NEXUS), codec.encodeContexts()); + } + + @Test + public void updateNotSentByNexusHandlerUsesWorkflowContext() { + codec.reset(); + Assert.assertEquals("hello", workflow.echo("hello")); + Assert.assertThrows(WorkflowUpdateException.class, () -> workflow.echo("fail")); + Assert.assertThrows(WorkflowUpdateException.class, () -> workflow.echo("reject")); + + Assert.assertFalse(codec.encodeContexts().contains(NEXUS)); + } + + /** Sends the echo update from inside a Nexus operation handler. */ + private String sendFromNexusHandler(String input) { + InternalNexusOperationContext nexusContext = + new InternalNexusOperationContext( + NAMESPACE, TASK_QUEUE, "endpoint", new NoopScope(), mock(WorkflowClient.class)); + nexusContext.setSerializationContext(NEXUS_CONTEXT); + CurrentNexusOperationContext.set(nexusContext); + try { + return workflow.echo(input); + } finally { + CurrentNexusOperationContext.unset(); + } + } + + private WorkflowExecutionHistory finishAndFetchHistory() { + workflow.finish(); + WorkflowStub stub = WorkflowStub.fromTyped(workflow); + stub.getResult(Void.class); + return testEnvironment + .getWorkflowClient() + .fetchHistory(workflowId, stub.getExecution().getRunId()); + } + + private String workflowContext() { + return ContextRecordingCodec.describe(new WorkflowSerializationContext(NAMESPACE, workflowId)); + } +} 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 index 51e18a7f86..2db5c243e0 100644 --- a/temporal-sdk/src/test/java/io/temporal/workflow/nexus/NexusSerializationContextPropagationTest.java +++ b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/NexusSerializationContextPropagationTest.java @@ -17,16 +17,21 @@ 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.history.v1.WorkflowExecutionUpdateAcceptedEventAttributes; +import io.temporal.api.history.v1.WorkflowExecutionUpdateCompletedEventAttributes; 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.UpdateOptions; import io.temporal.client.WorkflowClient; import io.temporal.client.WorkflowClientOptions; import io.temporal.client.WorkflowExecutionDescription; import io.temporal.client.WorkflowOptions; +import io.temporal.client.WorkflowStub; +import io.temporal.client.WorkflowUpdateStage; import io.temporal.common.RetryOptions; import io.temporal.common.WorkflowExecutionHistory; import io.temporal.failure.ApplicationFailure; @@ -44,6 +49,7 @@ import io.temporal.testing.internal.SDKTestWorkflowRule; import io.temporal.workflow.NexusOperationOptions; import io.temporal.workflow.NexusServiceOptions; +import io.temporal.workflow.UpdateMethod; import io.temporal.workflow.Workflow; import io.temporal.workflow.WorkflowInterface; import io.temporal.workflow.WorkflowMethod; @@ -61,9 +67,9 @@ 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. + * End-to-end coverage that workflows, workflow updates 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 @@ -80,7 +86,8 @@ public class NexusSerializationContextPropagationTest { @Rule public SDKTestWorkflowRule testWorkflowRule = SDKTestWorkflowRule.newBuilder() - .setWorkflowTypes(CallerWorkflowImpl.class, BackingWorkflowImpl.class) + .setWorkflowTypes( + CallerWorkflowImpl.class, BackingWorkflowImpl.class, UpdateTargetWorkflowImpl.class) .setActivityImplementations( new BackingActivityImpl(), new AsyncActivityImpl(), new StallingActivityImpl()) .setNexusServiceImplementation(new ContextServiceImpl()) @@ -197,6 +204,57 @@ public void continuedAsNewRunKeepsNexusContext() { CODEC.decodeContexts().contains(expected)); } + @Test + public void updateSentByHandlerUsesNexusContextForResult() { + String input = "update-" + RUN_ID; + UpdateTargetWorkflow target = + testWorkflowRule + .getWorkflowClient() + .newWorkflowStub( + UpdateTargetWorkflow.class, + WorkflowOptions.newBuilder() + .setWorkflowId(backingId(input)) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .build()); + WorkflowClient.start(target::execute); + + Assert.assertEquals("hello " + input, newCaller().execute("startUpdate", input)); + + String expected = describe("startUpdate"); + WorkflowStub targetStub = WorkflowStub.fromTyped(target); + WorkflowExecutionHistory history = testWorkflowRule.getExecutionHistory(backingId(input)); + WorkflowExecutionUpdateAcceptedEventAttributes accepted = null; + WorkflowExecutionUpdateCompletedEventAttributes completed = null; + for (HistoryEvent event : history.getEvents()) { + if (event.hasWorkflowExecutionUpdateAcceptedEventAttributes()) { + accepted = event.getWorkflowExecutionUpdateAcceptedEventAttributes(); + } else if (event.hasWorkflowExecutionUpdateCompletedEventAttributes()) { + completed = event.getWorkflowExecutionUpdateCompletedEventAttributes(); + } + } + Assert.assertNotNull("expected an accepted update", accepted); + Assert.assertNotNull("expected a completed update", completed); + Assert.assertEquals( + propagated("startUpdate"), + accepted.getAcceptedRequest().getPropagatedNexusSerializationContext()); + // Only the update's result, which is returned to the Nexus caller, uses the Nexus context. + Assert.assertEquals( + workflowContext(backingId(input)), + contextOf(accepted.getAcceptedRequest().getInput().getArgs().getPayloads(0))); + Assert.assertEquals(expected, contextOf(completed.getOutcome().getSuccess().getPayloads(0))); + + // A handle obtained by ID gets the context from the server when polling the outcome. + CODEC.reset(); + String result = targetStub.getUpdateHandle(backingId(input), String.class).getResult(); + Assert.assertEquals("hello " + input, result); + Assert.assertTrue( + "update handle should decode with the Nexus context, saw " + CODEC.decodeContexts(), + CODEC.decodeContexts().contains(expected)); + + target.finish(); + targetStub.getResult(Void.class); + } + @Test public void activityStartedByHandlerUsesNexusContextForResult() throws Exception { String input = "activity-" + RUN_ID; @@ -381,6 +439,9 @@ public interface ContextService { @Operation String startActivity(String input); + @Operation + String startUpdate(String input); + @Operation String startAsyncActivity(String input); @@ -456,6 +517,22 @@ public OperationHandler startStallingActivity() { .setRetryOptions(RetryOptions.newBuilder().setMaximumAttempts(1).build()) .build())); } + + @OperationImpl + public OperationHandler startUpdate() { + return TemporalOperationHandler.create( + (context, client, input) -> + client.startWorkflowUpdate( + UpdateTargetWorkflow.class, + backingId(input), + UpdateTargetWorkflow::echo, + input, + UpdateOptions.newBuilder(String.class) + .setUpdateName("echo") + .setUpdateId(backingId(input)) + .setWaitForStage(WorkflowUpdateStage.ACCEPTED) + .build())); + } } @WorkflowInterface @@ -480,6 +557,8 @@ public String execute(String operation, String input) { switch (operation) { case "startWorkflow": return service.startWorkflow(input); + case "startUpdate": + return service.startUpdate(input); case "startAsyncActivity": return service.startAsyncActivity(input); case "startStallingActivity": @@ -518,6 +597,37 @@ public String execute(String input) { } } + @WorkflowInterface + public interface UpdateTargetWorkflow { + @WorkflowMethod + void execute(); + + @UpdateMethod + String echo(String input); + + @UpdateMethod + void finish(); + } + + public static class UpdateTargetWorkflowImpl implements UpdateTargetWorkflow { + private boolean finished; + + @Override + public void execute() { + Workflow.await(() -> finished); + } + + @Override + public String echo(String input) { + return "hello " + input; + } + + @Override + public void finish() { + finished = true; + } + } + @ActivityInterface public interface BackingActivity { @ActivityMethod