#347

Committed 31 Jul 2026 06:59PM UTC coverage: 68.148% (-0.03%) from 68.181%

Build # #347

Build Type

push

github

Committed by

web-flow

Commit Message

NEXUS-485: Support Workflow Update as a Nexus Operation (#2945)

* NEXUS-485: Support Workflow Update as a Nexus Operation

* address comments, change signatures to newer

* address comments 2: add all workflow exec overloads

* address comments: log failed

Coverage Stats

7166 of 12554 branches covered (57.08%)

Branch coverage included in aggregate %.

118 of 296 new or added lines in 11 files covered. (39.86%)

28 existing lines in 7 files now uncovered.

29722 of 41575 relevant lines covered (71.49%)

0.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

69.23

/temporal-sdk/src/main/java/io/temporal/nexus/TemporalOperationHandler.java

package io.temporal.nexus;

import io.nexusrpc.OperationException;
import io.nexusrpc.handler.*;
import io.temporal.client.WorkflowClient;
import io.temporal.common.Experimental;
import io.temporal.internal.nexus.CurrentNexusOperationContext;
import io.temporal.internal.nexus.InternalNexusOperationContext;
import io.temporal.internal.nexus.OperationToken;
import io.temporal.internal.nexus.OperationTokenUtil;

/**
 * Generic Nexus operation handler backed by Temporal. Implements {@link OperationHandler} and
 * provides a composable way to map Temporal operations (start workflow, etc.) to Nexus operations.
 *
 * <p>Usage example:
 *
 * <pre>{@code
 * @OperationImpl
 * public OperationHandler<TransferInput, TransferResult> startTransfer() {
 *   return TemporalOperationHandler.create((context, client, input) -> {
 *     return client.startWorkflow(
 *         TransferWorkflow.class,
 *         TransferWorkflow::transfer, input.getFromAccount(), input.getToAccount(),
 *         WorkflowOptions.newBuilder()
 *             .setWorkflowId("transfer-" + input.getTransferId())
 *             .build());
 *   });
 * }
 * }</pre>
 *
 * <p>This class supports subclassing to customize cancel behavior. Override {@link
 * #cancelWorkflowRun} to change how workflow-run cancellations are handled. The {@link #start} and
 * {@link #cancel} methods should not be overridden — they contain the core dispatch logic.
 *
 * @param <T> the input type
 * @param <R> the result type
 */
@Experimental
public class TemporalOperationHandler<T, R> implements OperationHandler<T, R> {

  /**
   * Handler invoked when a Nexus start operation request is received.
   *
   * @param <T> the input type
   * @param <R> the result type
   */
  @FunctionalInterface
  public interface StartHandler<T, R> {
    TemporalOperationResult<R> apply(
        TemporalOperationStartContext context, TemporalNexusClient client, T input)
        throws OperationException;
  }

  private final StartHandler<T, R> startHandler;

  protected TemporalOperationHandler(StartHandler<T, R> startHandler) {
    this.startHandler = startHandler;
  }

  /**
   * Creates a {@link TemporalOperationHandler} from a start handler. Subclass and override {@link
   * #cancelWorkflowRun} to customize cancel behavior.
   *
   * @param startHandler the handler to invoke on start operation requests
   * @return an operation handler backed by the given start handler
   */
  public static <T, R> TemporalOperationHandler<T, R> create(StartHandler<T, R> startHandler) {
    return new TemporalOperationHandler<>(startHandler);
  }

  @Override
  public final OperationStartResult<R> start(
      OperationContext ctx, OperationStartDetails details, T input) throws OperationException {
    InternalNexusOperationContext nexusCtx = CurrentNexusOperationContext.get();
    TemporalNexusClient client =
        new TemporalNexusClientImpl(nexusCtx.getWorkflowClient(), ctx, details);

    TemporalOperationStartContext startContext = new TemporalOperationStartContext(ctx, details);
    TemporalOperationResult<R> result = startHandler.apply(startContext, client, input);

    if (result.isSync()) {
      return OperationStartResult.newSyncBuilder(result.getSyncResult()).build();
    } else if (result.isAsync()) {
      return OperationStartResult.<R>newAsyncBuilder(result.getAsyncOperationToken()).build();
    } else {
      throw new HandlerException(
          HandlerException.ErrorType.INTERNAL,
          new IllegalStateException("TemporalOperationResult must be either sync or async"));
    }
  }

  @Override
  public final void cancel(OperationContext ctx, OperationCancelDetails details) {
    OperationToken token;
    try {
      token = OperationTokenUtil.loadOperationToken(details.getOperationToken());
    } catch (IllegalArgumentException e) {
      throw new HandlerException(
          HandlerException.ErrorType.BAD_REQUEST, "failed to parse operation token", e);
    }

    TemporalOperationCancelContext cancelContext = new TemporalOperationCancelContext(ctx, details);
    switch (token.getType()) {
      case WORKFLOW_RUN:
        cancelWorkflowRun(cancelContext, new CancelWorkflowRunInput(token.getWorkflowId()));
        break;
      case WORKFLOW_UPDATE:
        cancelUpdateWorkflow(
            cancelContext,
            new CancelUpdateWorkflowInput(
                token.getWorkflowId(), token.getRunId(), token.getUpdateId()));
        break;
      default:
        throw new HandlerException(
            HandlerException.ErrorType.BAD_REQUEST,
            new IllegalArgumentException("unsupported operation token type: " + token.getType()));
    }
  }

  /**
   * Called when a cancel request is received for a workflow-run token (type=1). Override to
   * customize cancel behavior.
   *
   * <p>Default behavior: cancels the underlying workflow.
   *
   * @param context the cancel context
   * @param input describes the workflow run to cancel
   */
  protected void cancelWorkflowRun(
      TemporalOperationCancelContext context, CancelWorkflowRunInput input) {
    WorkflowClient client = CurrentNexusOperationContext.get().getWorkflowClient();
    client.newUntypedWorkflowStub(input.getWorkflowId()).cancel();
  }

  /**
   * Called when a cancel request is received for a workflow update token. Override to customize
   * cancel behavior.
   *
   * <p>Default behavior: not implemented. There is no server primitive to cancel an in-flight
   * workflow update.
   *
   * @param context the cancel context
   * @param input describes the update to cancel
   */
  protected void cancelUpdateWorkflow(
      TemporalOperationCancelContext context, CancelUpdateWorkflowInput input) {
    throw new HandlerException(
        HandlerException.ErrorType.NOT_IMPLEMENTED,
        new UnsupportedOperationException("cannot cancel an UpdateWorkflow operation"));
  }
}

1	package io.temporal.nexus;
2
3	import io.nexusrpc.OperationException;
4	import io.nexusrpc.handler.*;
5	import io.temporal.client.WorkflowClient;
6	import io.temporal.common.Experimental;
7	import io.temporal.internal.nexus.CurrentNexusOperationContext;
8	import io.temporal.internal.nexus.InternalNexusOperationContext;
9	import io.temporal.internal.nexus.OperationToken;
10	import io.temporal.internal.nexus.OperationTokenUtil;
11
12	/**
13	* Generic Nexus operation handler backed by Temporal. Implements {@link OperationHandler} and
14	* provides a composable way to map Temporal operations (start workflow, etc.) to Nexus operations.
15	*
16	* <p>Usage example:
17	*
18	* <pre>{@code
19	* @OperationImpl
20	* public OperationHandler<TransferInput, TransferResult> startTransfer() {
21	* return TemporalOperationHandler.create((context, client, input) -> {
22	* return client.startWorkflow(
23	* TransferWorkflow.class,
24	* TransferWorkflow::transfer, input.getFromAccount(), input.getToAccount(),
25	* WorkflowOptions.newBuilder()
26	* .setWorkflowId("transfer-" + input.getTransferId())
27	* .build());
28	* });
29	* }
30	* }</pre>
31	*
32	* <p>This class supports subclassing to customize cancel behavior. Override {@link
33	* #cancelWorkflowRun} to change how workflow-run cancellations are handled. The {@link #start} and
34	* {@link #cancel} methods should not be overridden — they contain the core dispatch logic.
35	*
36	* @param <T> the input type
37	* @param <R> the result type
38	*/
39	@Experimental
40	public class TemporalOperationHandler<T, R> implements OperationHandler<T, R> {
41
42	/**
43	* Handler invoked when a Nexus start operation request is received.
44	*
45	* @param <T> the input type
46	* @param <R> the result type
47	*/
48	@FunctionalInterface
49	public interface StartHandler<T, R> {
50	TemporalOperationResult<R> apply(
51	TemporalOperationStartContext context, TemporalNexusClient client, T input)
52	throws OperationException;
53	}
54
55	private final StartHandler<T, R> startHandler;
56
57	protected TemporalOperationHandler(StartHandler<T, R> startHandler) {	1✔
58	this.startHandler = startHandler;	1✔
59	}	1✔
60
61	/**
62	* Creates a {@link TemporalOperationHandler} from a start handler. Subclass and override {@link
63	* #cancelWorkflowRun} to customize cancel behavior.
64	*
65	* @param startHandler the handler to invoke on start operation requests
66	* @return an operation handler backed by the given start handler
67	*/
68	public static <T, R> TemporalOperationHandler<T, R> create(StartHandler<T, R> startHandler) {
69	return new TemporalOperationHandler<>(startHandler);	1✔
70	}
71
72	@Override
73	public final OperationStartResult<R> start(
74	OperationContext ctx, OperationStartDetails details, T input) throws OperationException {
75	InternalNexusOperationContext nexusCtx = CurrentNexusOperationContext.get();	1✔
76	TemporalNexusClient client =	1✔
77	new TemporalNexusClientImpl(nexusCtx.getWorkflowClient(), ctx, details);	1✔
78
79	TemporalOperationStartContext startContext = new TemporalOperationStartContext(ctx, details);	1✔
80	TemporalOperationResult<R> result = startHandler.apply(startContext, client, input);	1✔
81
82	if (result.isSync()) {	1✔
83	return OperationStartResult.newSyncBuilder(result.getSyncResult()).build();	1✔
84	} else if (result.isAsync()) {	1!
85	return OperationStartResult.<R>newAsyncBuilder(result.getAsyncOperationToken()).build();	1✔
86	} else {
87	throw new HandlerException(	×
88	HandlerException.ErrorType.INTERNAL,
89	new IllegalStateException("TemporalOperationResult must be either sync or async"));
90	}
91	}
92
93	@Override
94	public final void cancel(OperationContext ctx, OperationCancelDetails details) {
95	OperationToken token;
96	try {
97	token = OperationTokenUtil.loadOperationToken(details.getOperationToken());	1✔
98	} catch (IllegalArgumentException e) {	×
99	throw new HandlerException(	×
100	HandlerException.ErrorType.BAD_REQUEST, "failed to parse operation token", e);
101	}	1✔
102
103	TemporalOperationCancelContext cancelContext = new TemporalOperationCancelContext(ctx, details);	1✔
104	switch (token.getType()) {	1!
105	case WORKFLOW_RUN:
106	cancelWorkflowRun(cancelContext, new CancelWorkflowRunInput(token.getWorkflowId()));	1✔
107	break;	1✔
108	case WORKFLOW_UPDATE:
NEW 109	cancelUpdateWorkflow(	×
110	cancelContext,
111	new CancelUpdateWorkflowInput(
NEW 112	token.getWorkflowId(), token.getRunId(), token.getUpdateId()));	×
NEW 113	break;	×
114	default:
NEW 115	throw new HandlerException(	×
116	HandlerException.ErrorType.BAD_REQUEST,
NEW 117	new IllegalArgumentException("unsupported operation token type: " + token.getType()));	×
118	}
119	}	1✔
120
121	/**
122	* Called when a cancel request is received for a workflow-run token (type=1). Override to
123	* customize cancel behavior.
124	*
125	* <p>Default behavior: cancels the underlying workflow.
126	*
127	* @param context the cancel context
128	* @param input describes the workflow run to cancel
129	*/
130	protected void cancelWorkflowRun(
131	TemporalOperationCancelContext context, CancelWorkflowRunInput input) {
132	WorkflowClient client = CurrentNexusOperationContext.get().getWorkflowClient();	1✔
133	client.newUntypedWorkflowStub(input.getWorkflowId()).cancel();	1✔
134	}	1✔
135
136	/**
137	* Called when a cancel request is received for a workflow update token. Override to customize
138	* cancel behavior.
139	*
140	* <p>Default behavior: not implemented. There is no server primitive to cancel an in-flight
141	* workflow update.
142	*
143	* @param context the cancel context
144	* @param input describes the update to cancel
145	*/
146	protected void cancelUpdateWorkflow(
147	TemporalOperationCancelContext context, CancelUpdateWorkflowInput input) {
NEW 148	throw new HandlerException(	×
149	HandlerException.ErrorType.NOT_IMPLEMENTED,
150	new UnsupportedOperationException("cannot cancel an UpdateWorkflow operation"));
151	}
152	}

temporalio / sdk-java / #347

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous